[PR-verification] fix(low-code): support AsyncRetriever in DynamicSchemaLoader

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -2760,14 +2760,22 @@ def create_dynamic_schema_loader(
                     self._create_component_from_model(model=transformation_model, config=config)
                 )
         name = "dynamic_properties"
+        partition_router = self._build_stream_slicer_from_partition_router(model.retriever, config)
+        retriever_kwargs: Dict[str, Any] = {}
+        if isinstance(model.retriever, AsyncRetrieverModel):
+            if model.retriever.partition_router:
+                raise ValueError(
+                    "DynamicSchemaLoader does not support an AsyncRetriever with a partition_router: only the first record is read, so jobs for other partitions would be orphaned. Remove the partition_router from the schema loader's AsyncRetriever."
+                )
+            # create_async_retriever requires stream_slicer; passing it to a CustomRetriever would override its own
+            retriever_kwargs["stream_slicer"] = partition_router
         retriever = self._create_component_from_model(
             model=model.retriever,
             config=config,
             name=name,
             primary_key=None,
-            partition_router=self._build_stream_slicer_from_partition_router(
-                model.retriever, config
-            ),
+            partition_router=partition_router,
+            **retriever_kwargs,
             transformations=[],
             use_cache=True,
             log_formatter=(

airbyte_cdk/sources/declarative/retrievers/async_retriever.py

@@ -82,6 +82,12 @@ def read_records(
         records_schema: Mapping[str, Any],
         stream_slice: Optional[StreamSlice] = None,
     ) -> Iterable[StreamData]:
+        if stream_slice is None:
+            # Slice-less callers such as DynamicSchemaLoader read this way: create and poll the jobs here
+            for job_slice in self.stream_slicer.stream_slices():
+                yield from self.read_records(records_schema, job_slice)
+            return
+
         # emit the slice_descriptor log message, for connector builder TestRead
         yield self.slice_logger.create_slice_log_message(stream_slice.cursor_slice)  # type: ignore
 

airbyte_cdk/sources/declarative/schema/dynamic_schema_loader.py

@@ -10,6 +10,7 @@
 import dpath
 from typing_extensions import deprecated
 
+from airbyte_cdk.models import AirbyteMessage, Type
 from airbyte_cdk.sources.declarative.extractors.record_filter import RecordFilter
 from airbyte_cdk.sources.declarative.interpolation.interpolated_boolean import InterpolatedBoolean
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
@@ -120,6 +121,10 @@ def _update_pointer(
 class DynamicSchemaLoader(SchemaLoader):
     """
     Dynamically loads a JSON Schema by extracting data from retrieved records.
+
+    With an AsyncRetriever, each schema load creates one async job (retried on failure) and stops at
+    the first record, so the job's delete endpoint is not called. During a read the schema loads on
+    the first partition read and its job takes a slot from max_concurrent_async_job_count.
     """
 
     retriever: Retriever
@@ -134,7 +139,15 @@ def get_json_schema(self) -> Mapping[str, Any]:
         Constructs a JSON Schema based on retrieved data.
         """
         properties = {}
-        retrieved_record = next(self.retriever.read_records({}), None)  # type: ignore[call-overload] # read_records return Iterable data type
+        retrieved_record = next(
+            (
+                record
+                for record in self.retriever.read_records({})
+                # AsyncRetriever emits a slice log message before its records
+                if not (isinstance(record, AirbyteMessage) and record.type == Type.LOG)
+            ),
+            None,
+        )
 
         raw_schema = (
             self._extract_data(

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -79,6 +79,9 @@
 from airbyte_cdk.sources.declarative.models import DeclarativeStream as DeclarativeStreamModel
 from airbyte_cdk.sources.declarative.models import DefaultPaginator as DefaultPaginatorModel
 from airbyte_cdk.sources.declarative.models import DpathExtractor as DpathExtractorModel
+from airbyte_cdk.sources.declarative.models import (
+    DynamicSchemaLoader as DynamicSchemaLoaderModel,
+)
 from airbyte_cdk.sources.declarative.models import FileUploader as FileUploaderModel
 from airbyte_cdk.sources.declarative.models import (
     GroupingPartitionRouter as GroupingPartitionRouterModel,
@@ -205,7 +208,11 @@
     PageSizeReduction,
     PageSizeResetPolicy,
 )
-from airbyte_cdk.sources.declarative.schema import InlineSchemaLoader, JsonFileSchemaLoader
+from airbyte_cdk.sources.declarative.schema import (
+    DynamicSchemaLoader,
+    InlineSchemaLoader,
+    JsonFileSchemaLoader,
+)
 from airbyte_cdk.sources.declarative.schema.caching_schema_loader_decorator import (
     CachingSchemaLoaderDecorator,
 )
@@ -5124,6 +5131,137 @@ def test_create_async_retriever():
     assert download_retriever_record_selector.schema_normalization._config.name == "NoTransform"
 
 
+def test_dynamic_schema_loader_with_async_retriever():
+    """
+    Verifies that DynamicSchemaLoader can be created with an AsyncRetriever without raising
+    TypeError due to missing stream_slicer kwarg.
+    Regression test for https://github.com/airbytehq/airbyte-python-cdk/issues/766
+    """
+    content = """
+stream_with_dynamic_schema:
+  type: DeclarativeStream
+  name: test_stream
+  schema_loader:
+    type: DynamicSchemaLoader
+    retriever:
+      type: AsyncRetriever
+      record_selector:
+        type: RecordSelector
+        extractor:
+          type: DpathExtractor
+          field_path: ["data"]
+      status_mapping:
+        failed:
+          - Error
+        running:
+          - Pending
+        completed:
+          - Success
+        timeout: []
+      status_extractor:
+        type: DpathExtractor
+        field_path:
+          - status
+      download_target_extractor:
+        type: DpathExtractor
+        field_path:
+          - download_url
+      creation_requester:
+        type: HttpRequester
+        url_base: https://api.test.com
+        path: /reports/create
+        http_method: POST
+      polling_requester:
+        type: HttpRequester
+        url_base: https://api.test.com
+        path: /reports/status
+        http_method: GET
+      download_requester:
+        type: HttpRequester
+        url_base: "{{download_target}}"
+        http_method: GET
+    schema_type_identifier:
+      type: SchemaTypeIdentifier
+      key_pointer:
+        - name
+      type_pointer:
+        - type
+  retriever:
+    type: SimpleRetriever
+    requester:
+      type: HttpRequester
+      url_base: https://api.test.com
+      path: /data
+      http_method: GET
+    record_selector:
+      type: RecordSelector
+      extractor:
+        type: DpathExtractor
+        field_path: []
+    paginator:
+      type: NoPagination
+  $parameters:
+    name: "test_stream"
+    primary_key: "id"
+"""
+    parsed_manifest = YamlDeclarativeSource._parse(content)
+    resolved_manifest = resolver.preprocess_manifest(parsed_manifest)
+    stream_manifest = transformer.propagate_types_and_parameters(
+        "", resolved_manifest["stream_with_dynamic_schema"], {}
+    )
+
+    stream = factory.create_component(
+        model_type=DeclarativeStreamModel,
+        component_definition=stream_manifest,
+        config=input_config,
+    )
+
+    assert isinstance(stream, DefaultStream)
+    schema_loader = get_schema_loader(stream)
+    assert isinstance(schema_loader, DynamicSchemaLoader)
+    assert isinstance(schema_loader.retriever, AsyncRetriever)
+
+
+def test_dynamic_schema_loader_keeps_custom_retriever_stream_slicer():
+    content = """
+schema_loader:
+  type: DynamicSchemaLoader
+  retriever:
+    type: CustomRetriever
+    class_name: unit_tests.sources.declarative.parsers.testing_components.TestingCustomRetriever
+    stream_slicer:
+      type: ListPartitionRouter
+      cursor_field: p
+      values: ["a"]
+    requester:
+      type: HttpRequester
+      url: https://api.test.com/schema
+    record_selector:
+      type: RecordSelector
+      extractor:
+        type: DpathExtractor
+        field_path: []
+  schema_type_identifier:
+    type: SchemaTypeIdentifier
+    key_pointer:
+      - name
+"""
+    parsed_manifest = YamlDeclarativeSource._parse(content)
+    resolved_manifest = resolver.preprocess_manifest(parsed_manifest)
+    schema_loader_manifest = transformer.propagate_types_and_parameters(
+        "", resolved_manifest["schema_loader"], {}
+    )
+
+    schema_loader = factory.create_component(
+        model_type=DynamicSchemaLoaderModel,
+        component_definition=schema_loader_manifest,
+        config=input_config,
+    )
+
+    assert isinstance(schema_loader.retriever, TestingCustomRetriever)
+    assert isinstance(schema_loader.retriever.stream_slicer, ListPartitionRouter)
+
+
 def test_api_budget():
     manifest = {
         "type": "DeclarativeSource",