feat(capabilities): support for native query connections

src/_example/django/django_demo/.forestadmin-schema.json

@@ -2726,7 +2726,7 @@
                     "enums": null,
                     "field": "body",
                     "inverseOf": null,
-                    "isFilterable": false,
+                    "isFilterable": true,
                     "isPrimaryKey": false,
                     "isReadOnly": false,
                     "isRequired": false,
@@ -2740,7 +2740,7 @@
                     "enums": null,
                     "field": "email",
                     "inverseOf": null,
-                    "isFilterable": false,
+                    "isFilterable": true,
                     "isPrimaryKey": false,
                     "isReadOnly": false,
                     "isRequired": false,
@@ -2768,7 +2768,7 @@
                     "enums": null,
                     "field": "name",
                     "inverseOf": null,
-                    "isFilterable": false,
+                    "isFilterable": true,
                     "isPrimaryKey": false,
                     "isReadOnly": false,
                     "isRequired": false,
@@ -3486,7 +3486,7 @@
                     "enums": null,
                     "field": "body",
                     "inverseOf": null,
-                    "isFilterable": false,
+                    "isFilterable": true,
                     "isPrimaryKey": false,
                     "isReadOnly": false,
                     "isRequired": false,
@@ -3514,7 +3514,7 @@
                     "enums": null,
                     "field": "title",
                     "inverseOf": null,
-                    "isFilterable": false,
+                    "isFilterable": true,
                     "isPrimaryKey": false,
                     "isReadOnly": false,
                     "isRequired": false,

src/_example/django/django_demo/app/forest/custom_datasources/typicode.py

@@ -107,14 +107,17 @@ def __init__(self, datasource: Datasource[Self]):
                 "name": {
                     "type": FieldType.COLUMN,
                     "column_type": "String",
+                    "filter_operators": set([Operator.EQUAL]),
                 },
                 "email": {
                     "type": FieldType.COLUMN,
                     "column_type": "String",
+                    "filter_operators": set([Operator.EQUAL]),
                 },
                 "body": {
                     "type": FieldType.COLUMN,
                     "column_type": "String",
+                    "filter_operators": set([Operator.EQUAL]),
                 },
             }
         )
@@ -140,10 +143,12 @@ def __init__(self, datasource: Datasource[Self]):
                 "title": {
                     "type": FieldType.COLUMN,
                     "column_type": "String",
+                    "filter_operators": set([Operator.EQUAL]),
                 },
                 "body": {
                     "type": FieldType.COLUMN,
                     "column_type": "String",
+                    "filter_operators": set([Operator.EQUAL]),
                 },
             }
         )

src/_example/django/django_demo/app/forest/customer.py

@@ -69,7 +69,7 @@ async def get_customer_spending_values(records: List[RecordsDataAlias], context:
 
 def customer_full_name() -> ComputedDefinition:
     async def _get_customer_fullname_values(records: List[RecordsDataAlias], context: CollectionCustomizationContext):
-        return [f"{record['first_name']} - {record['last_name']}" for record in records]
+        return [f"{record.get('first_name', '')} - {record.get('last_name', '')}" for record in records]
 
     return {
         "column_type": "String",

src/_example/django/django_demo/app/forest_admin.py

@@ -50,9 +50,16 @@
 
 def customize_forest(agent: DjangoAgent):
     # customize_forest_logging()
-    agent.add_datasource(DjangoDatasource(support_polymorphic_relations=True))
+
+    agent.add_datasource(
+        DjangoDatasource(
+            support_polymorphic_relations=True, live_query_connection={"django": "default", "dj_sqlachemy": "other"}
+        )
+    )
     agent.add_datasource(TypicodeDatasource())
-    agent.add_datasource(SqlAlchemyDatasource(Base, DB_URI))
+    agent.add_datasource(
+        SqlAlchemyDatasource(Base, DB_URI, live_query_connection="sqlalchemy"),
+    )
 
     agent.customize_collection("address").add_segment("France", segment_addr_fr("address"))
     agent.customize_collection("app_address").add_segment("France", segment_addr_fr("app_address"))

src/_example/django/django_demo/django_demo/settings.py

@@ -123,10 +123,10 @@
         "NAME": os.path.join(BASE_DIR, "db.sqlite3"),
         "ATOMIC_REQUESTS": True,
     },
-    # "other": {
-    #     "ENGINE": "django.db.backends.sqlite3",
-    #     "NAME": os.path.join(BASE_DIR, "db_flask_example.sqlite"),
-    # },
+    "other": {
+        "ENGINE": "django.db.backends.sqlite3",
+        "NAME": os.path.join(BASE_DIR, "db_sqlalchemy.sql"),
+    },
 }
 
 DATABASE_ROUTERS = ["django_demo.db_router.DBRouter"]

src/agent_toolkit/forestadmin/agent_toolkit/agent.py

@@ -9,6 +9,7 @@
 from forestadmin.agent_toolkit.resources.collections.charts_datasource import ChartsDatasourceResource
 from forestadmin.agent_toolkit.resources.collections.crud import CrudResource
 from forestadmin.agent_toolkit.resources.collections.crud_related import CrudRelatedResource
+from forestadmin.agent_toolkit.resources.collections.native_query import NativeQueryResource
 from forestadmin.agent_toolkit.resources.collections.stats import StatsResource
 from forestadmin.agent_toolkit.resources.security.resources import Authentication
 from forestadmin.agent_toolkit.services.permissions.ip_whitelist_service import IpWhiteListService
@@ -37,6 +38,7 @@ class Resources(TypedDict):
     actions: ActionResource
     collection_charts: ChartsCollectionResource
     datasource_charts: ChartsDatasourceResource
+    native_query: NativeQueryResource
 
 
 class Agent:
@@ -73,10 +75,13 @@ def __del__(self):
     async def __mk_resources(self):
         self._resources: Resources = {
             "capabilities": CapabilitiesResource(
-                await self.customizer.get_datasource(), self._ip_white_list_service, self.options
+                self.customizer.composite_datasource,
+                self._ip_white_list_service,
+                self.options,
             ),
             "authentication": Authentication(self._ip_white_list_service, self.options),
             "crud": CrudResource(
+                self.customizer.composite_datasource,
                 await self.customizer.get_datasource(),
                 self._permission_service,
                 self._ip_white_list_service,
@@ -112,14 +117,23 @@ async def __mk_resources(self):
                 self._ip_white_list_service,
                 self.options,
             ),
+            "native_query": NativeQueryResource(
+                self.customizer.composite_datasource,
+                await self.customizer.get_datasource(),
+                self._permission_service,
+                self._ip_white_list_service,
+                self.options,
+            ),
         }
 
     async def get_resources(self):
         if self._resources is None:
             await self.__mk_resources()
         return self._resources
 
-    def add_datasource(self, datasource: Datasource[BoundCollection], options: Optional[DataSourceOptions] = None):
+    def add_datasource(
+        self, datasource: Datasource[BoundCollection], options: Optional[DataSourceOptions] = None
+    ) -> Self:
         """Add a datasource
 
         Args:
@@ -130,6 +144,7 @@ def add_datasource(self, datasource: Datasource[BoundCollection], options: Optio
             options = {}
         self.customizer.add_datasource(datasource, options)
         self._resources = None
+        return self
 
     def use(self, plugin: type, options: Optional[Dict] = {}) -> Self:
         """Load a plugin across all collections

src/agent_toolkit/forestadmin/agent_toolkit/resources/capabilities.py

@@ -7,6 +7,7 @@
 from forestadmin.agent_toolkit.services.permissions.ip_whitelist_service import IpWhiteListService
 from forestadmin.agent_toolkit.utils.context import HttpResponseBuilder, Request, RequestMethod, Response
 from forestadmin.agent_toolkit.utils.forest_schema.generator_field import SchemaFieldGenerator
+from forestadmin.datasource_toolkit.datasource_customizer.datasource_composite import CompositeDatasource
 from forestadmin.datasource_toolkit.datasource_customizer.datasource_customizer import DatasourceCustomizer
 from forestadmin.datasource_toolkit.exceptions import BusinessError
 from forestadmin.datasource_toolkit.interfaces.fields import Column, is_column
@@ -19,12 +20,12 @@
 class CapabilitiesResource(IpWhitelistResource):
     def __init__(
         self,
-        datasource: DatasourceAlias,
+        composite_datasource: CompositeDatasource,
         ip_white_list_service: IpWhiteListService,
         options: Options,
     ):
         super().__init__(ip_white_list_service, options)
-        self.datasource = datasource
+        self.composite_datasource: CompositeDatasource = composite_datasource
 
     @ip_white_list
     async def dispatch(self, request: Request, method_name: LiteralMethod) -> Response:
@@ -40,14 +41,18 @@ async def dispatch(self, request: Request, method_name: LiteralMethod) -> Respon
     @check_method(RequestMethod.POST)
     @authenticate
     async def capabilities(self, request: Request) -> Response:
-        ret = {"collections": []}
+        ret = {"collections": [], "nativeQueryConnections": []}
         requested_collections = request.body.get("collectionNames", [])
         for collection_name in requested_collections:
             ret["collections"].append(self._get_collection_capability(collection_name))
+
+        ret["nativeQueryConnections"] = [
+            {"name": connection} for connection in self.composite_datasource.get_native_query_connections()
+        ]
         return HttpResponseBuilder.build_success_response(ret)
 
     def _get_collection_capability(self, collection_name: str) -> Dict[str, Any]:
-        collection = self.datasource.get_collection(collection_name)
+        collection = self.composite_datasource.get_collection(collection_name)
         fields = []
         for field_name, field_schema in collection.schema["fields"].items():
             if is_column(field_schema):

src/agent_toolkit/forestadmin/agent_toolkit/resources/collections/crud.py

@@ -1,8 +1,9 @@
 import asyncio
-from typing import Any, Awaitable, Dict, List, Literal, Tuple, Union, cast
+from typing import Any, Awaitable, Dict, List, Literal, Optional, Tuple, Union, cast
 from uuid import UUID
 
 from forestadmin.agent_toolkit.forest_logger import ForestLogger
+from forestadmin.agent_toolkit.options import Options
 from forestadmin.agent_toolkit.resources.collections.base_collection_resource import BaseCollectionResource
 from forestadmin.agent_toolkit.resources.collections.decorators import (
     authenticate,
@@ -21,15 +22,21 @@
     parse_timezone,
 )
 from forestadmin.agent_toolkit.resources.collections.requests import RequestCollection, RequestCollectionException
+from forestadmin.agent_toolkit.resources.context_variable_injector_mixin import ContextVariableInjectorResourceMixin
+from forestadmin.agent_toolkit.services.permissions.ip_whitelist_service import IpWhiteListService
+from forestadmin.agent_toolkit.services.permissions.permission_service import PermissionService
 from forestadmin.agent_toolkit.services.serializers import add_search_metadata
 from forestadmin.agent_toolkit.services.serializers.json_api import JsonApiException, JsonApiSerializer
 from forestadmin.agent_toolkit.utils.context import HttpResponseBuilder, Request, RequestMethod, Response, User
 from forestadmin.agent_toolkit.utils.csv import Csv, CsvException
 from forestadmin.agent_toolkit.utils.id import unpack_id
+from forestadmin.agent_toolkit.utils.sql_query_checker import SqlQueryChecker
 from forestadmin.datasource_toolkit.collections import Collection
 from forestadmin.datasource_toolkit.datasource_customizer.collection_customizer import CollectionCustomizer
-from forestadmin.datasource_toolkit.datasources import DatasourceException
-from forestadmin.datasource_toolkit.exceptions import ForbiddenError
+from forestadmin.datasource_toolkit.datasource_customizer.datasource_composite import CompositeDatasource
+from forestadmin.datasource_toolkit.datasource_customizer.datasource_customizer import DatasourceCustomizer
+from forestadmin.datasource_toolkit.datasources import Datasource, DatasourceException
+from forestadmin.datasource_toolkit.exceptions import ForbiddenError, NativeQueryException
 from forestadmin.datasource_toolkit.interfaces.fields import (
     ManyToOne,
     OneToOne,
@@ -55,14 +62,26 @@
 from forestadmin.datasource_toolkit.interfaces.query.projections.factory import ProjectionFactory
 from forestadmin.datasource_toolkit.interfaces.records import CompositeIdAlias, RecordsDataAlias
 from forestadmin.datasource_toolkit.utils.collections import CollectionUtils
+from forestadmin.datasource_toolkit.utils.records import RecordUtils
 from forestadmin.datasource_toolkit.utils.schema import SchemaUtils
 from forestadmin.datasource_toolkit.validations.field import FieldValidatorException
 from forestadmin.datasource_toolkit.validations.records import RecordValidator, RecordValidatorException
 
 LiteralMethod = Literal["list", "count", "add", "get", "delete_list", "csv"]
 
 
-class CrudResource(BaseCollectionResource):
+class CrudResource(BaseCollectionResource, ContextVariableInjectorResourceMixin):
+    def __init__(
+        self,
+        datasource_composite: CompositeDatasource,
+        datasource: Union[Datasource, DatasourceCustomizer],
+        permission: PermissionService,
+        ip_white_list_service: IpWhiteListService,
+        options: Options,
+    ):
+        self._datasource_composite = datasource_composite
+        super().__init__(datasource, permission, ip_white_list_service, options)
+
     @ip_white_list
     async def dispatch(self, request: Request, method_name: LiteralMethod) -> Response:
         method = getattr(self, method_name)
@@ -159,6 +178,9 @@ async def list(self, request: RequestCollection) -> Response:
         scope_tree = await self.permission.get_scope(request.user, request.collection)
         try:
             paginated_filter = build_paginated_filter(request, scope_tree)
+            condition_tree = await self._handle_live_query_segment(request, paginated_filter.condition_tree)
+            paginated_filter = paginated_filter.override({"condition_tree": condition_tree})
+
         except FilterException as e:
             ForestLogger.log("exception", e)
             return HttpResponseBuilder.build_client_error_response([e])
@@ -191,6 +213,8 @@ async def csv(self, request: RequestCollection) -> Response:
         scope_tree = await self.permission.get_scope(request.user, request.collection)
         try:
             paginated_filter = build_paginated_filter(request, scope_tree)
+            condition_tree = await self._handle_live_query_segment(request, paginated_filter.condition_tree)
+            paginated_filter = paginated_filter.override({"condition_tree": condition_tree})
             paginated_filter.page = None
         except FilterException as e:
             ForestLogger.log("exception", e)
@@ -220,9 +244,12 @@ async def count(self, request: RequestCollection) -> Response:
             return HttpResponseBuilder.build_success_response({"meta": {"count": "deactivated"}})
 
         scope_tree = await self.permission.get_scope(request.user, request.collection)
-        filter = build_filter(request, scope_tree)
+        filter_ = build_filter(request, scope_tree)
+        filter_ = filter_.override(
+            {"condition_tree": await self._handle_live_query_segment(request, filter_.condition_tree)}
+        )
         aggregation = Aggregation({"operation": "Count"})
-        result = await request.collection.aggregate(request.user, filter, aggregation)
+        result = await request.collection.aggregate(request.user, filter_, aggregation)
         try:
             count = result[0]["value"]
         except IndexError:
@@ -424,3 +451,33 @@ def _serialize_records_with_relationships(
 
         schema = JsonApiSerializer.get(collection)
         return schema(projections=projection).dump(records if many is True else records[0], many=many)
+
+    async def _handle_live_query_segment(
+        self, request: RequestCollection, condition_tree: Optional[ConditionTree]
+    ) -> Optional[ConditionTree]:
+        if request.query.get("segmentQuery") is not None:
+            if request.query.get("connectionName") in ["", None]:
+                raise NativeQueryException("Missing native query connection attribute")
+
+            await self.permission.can_live_query_segment(request)
+            SqlQueryChecker.check_query(request.query["segmentQuery"])
+            variables = await self.inject_and_get_context_variables_in_live_query_segment(request)
+            native_query_result = await self._datasource_composite.execute_native_query(
+                request.query["connectionName"], request.query["segmentQuery"], variables
+            )
+
+            pk_field = SchemaUtils.get_primary_keys(request.collection.schema)[0]
+            if len(native_query_result) > 0 and pk_field not in native_query_result[0]:
+                raise NativeQueryException(f"Live query must return the primary key field ('{pk_field}').")
+
+            trees = []
+            if condition_tree:
+                trees.append(condition_tree)
+            trees.append(
+                ConditionTreeFactory.match_ids(
+                    request.collection.schema,
+                    [RecordUtils.get_primary_key(request.collection.schema, r) for r in native_query_result],
+                )
+            )
+            return ConditionTreeFactory.intersect(trees)
+        return condition_tree