feat: BI-6681 add connection source listing options handler (#1294)

* feat: BI-6681 add connection source listing options handler

* fix

* fix test

* put class names in order

Author: KonstantAnxiety

lib/dl_api_lib/dl_api_lib/app/control_api/resources/connections.py

@@ -28,6 +28,7 @@
     ConnectionDBNamesResponseSchema,
     ConnectionExportResponseSchema,
     ConnectionImportRequestSchema,
+    ConnectionInfoSourceListingOptionsResponseSchema,
     ConnectionInfoSourceSchemaQuerySchema,
     ConnectionInfoSourceSchemaResponseSchema,
     ConnectionItemQuerySchema,
@@ -56,6 +57,7 @@
 from dl_core.us_connection_base import (
     ConnectionBase,
     DataSourceTemplate,
+    ListingOptions,
 )
 
 
@@ -406,6 +408,41 @@ def get(self, connection_id: str) -> dict[str, list[str]]:
         }
 
 
+@ns.route("/<connection_id>/info/source_listing_options")
+class ConnectionInfoSourceListingOptions(BIResource):
+    @schematic_request(
+        ns=ns,
+        responses={
+            200: ("Success", ConnectionInfoSourceListingOptionsResponseSchema()),
+            400: ("Failed", BadRequestResponseSchema()),
+        },
+    )
+    def get(self, connection_id: str) -> dict:
+        connection = self.get_us_manager().get_by_id(connection_id, expected_type=ConnectionBase)
+
+        service_registry = self.get_service_registry()
+        localizer = service_registry.get_localizer()
+
+        # It does not matter what options we provide if the user does not have sufficient permissions,
+        # because the listing itself will not attempt to list actual DB sources, see `/info/sources`
+        listing_options = ListingOptions(
+            supports_source_search=False,
+            supports_source_pagination=False,
+            supports_db_name_listing=False,
+            db_name_required_for_search=False,
+            db_name_label=None,
+        )
+        try:
+            need_permission_on_entry(connection, USPermissionKind.read)
+        except USPermissionRequired:
+            pass
+        else:
+            localizer = service_registry.get_localizer()
+            listing_options = connection.get_listing_options(localizer)
+
+        return {"source_listing": listing_options}
+
+
 @ns.route("/<connection_id>/info/sources")
 class ConnectionInfoSources(BIResource):
     @schematic_request(

lib/dl_api_lib/dl_api_lib/schemas/connection.py

@@ -19,6 +19,7 @@
     RawSchemaColumnSchema,
 )
 from dl_api_lib.schemas.main import NotificationContentSchema
+from dl_api_lib.schemas.options import SourceListingSchema
 from dl_constants.enums import ConnectionType as CT
 from dl_core.us_connection_base import (
     ConnectionBase,
@@ -45,6 +46,10 @@ class ConnectionInfoSourceSchemaQuerySchema(BaseSchema):
     source = ma_fields.Nested(DataSourceSchema, required=True)
 
 
+class ConnectionInfoSourceListingOptionsResponseSchema(BaseSchema):
+    source_listing = ma_fields.Nested(SourceListingSchema)
+
+
 class ConnectionSourceTemplatesResponseSchema(BaseSchema):
     # # Complete:
     # sources = ma_fields.Nested(DataSourceTemplateResponseSchema, many=True)

lib/dl_api_lib_testing/dl_api_lib_testing/connector/connection_suite.py

@@ -286,6 +286,21 @@ def test_connection_sources(
         assert "sources" in resp_data, resp_data
         assert isinstance(resp_data["sources"], list), resp_data
 
+    def test_connection_source_listing_options(
+        self,
+        control_api_sync_client: SyncHttpClientBase,
+        saved_connection_id: str,
+        bi_headers: dict[str, str] | None,
+    ) -> None:
+        resp = control_api_sync_client.get(
+            url=f"/api/v1/connections/{saved_connection_id}/info/source_listing_options",
+            headers=bi_headers,
+        )
+        assert resp.status_code == 200, resp.json
+        resp_data = resp.json
+        assert "source_listing" in resp_data, resp_data
+        assert isinstance(resp_data["source_listing"], dict), resp_data
+
     def test_connection_description(
         self,
         control_api_sync_client: SyncHttpClientBase,