fix

Author: bazarnov

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1779,6 +1779,8 @@ definitions:
           - stream_interval
           - stream_partition
           - stream_slice
+          - creation_response
+          - download_target
         examples:
           - "/products"
           - "/quotes/{{ stream_partition['id'] }}/quote_line_groups"
@@ -3223,7 +3225,7 @@ definitions:
       - polling_requester
       - download_requester
       - status_extractor
-      - urls_extractor
+      - download_target_extractor
     properties:
       type:
         type: string
@@ -3240,7 +3242,7 @@ definitions:
         anyOf:
           - "$ref": "#/definitions/CustomRecordExtractor"
           - "$ref": "#/definitions/DpathExtractor"
-      urls_extractor:
+      download_target_extractor:
         description: Responsible for fetching the final result `urls` provided by the completed / finished / ready async job.
         anyOf:
           - "$ref": "#/definitions/CustomRecordExtractor"
@@ -3667,6 +3669,16 @@ interpolation:
             self: https://api.sendgrid.com/v3/marketing/lists?page_size=1&page_token=
             next: https://api.sendgrid.com/v3/marketing/lists?page_size=1&page_token=0236d6d2
             count: 82
+    - title: creation_response
+      description: The response received from the creation_requester in the AsyncRetriever component.
+      type: object
+      examples:
+        - id: "1234"
+    - title: download_target
+      description: The `URL` received from the polling_requester in the AsyncRetriever with jobStatus as `COMPLETED`.
+      type: string
+      examples:
+        - url: "https://api.sendgrid.com/v3/marketing/lists?page_size=1&page_token=0236d6d2&filename=xxx_yyy_zzz.csv"
     - title: stream_interval
       description: The current stream interval being processed. The keys are defined by the incremental sync component. Default keys are `start_time` and `end_time`.
       type: object

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -2263,7 +2263,7 @@ class AsyncRetriever(BaseModel):
     status_extractor: Union[CustomRecordExtractor, DpathExtractor] = Field(
         ..., description="Responsible for fetching the actual status of the async job."
     )
-    urls_extractor: Union[CustomRecordExtractor, DpathExtractor] = Field(
+    download_target_extractor: Union[CustomRecordExtractor, DpathExtractor] = Field(
         ...,
         description="Responsible for fetching the final result `urls` provided by the completed / finished / ready async job.",
     )

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -2757,8 +2757,8 @@ def _get_download_retriever() -> SimpleRetrieverTestReadDecorator | SimpleRetrie
         status_extractor = self._create_component_from_model(
             model=model.status_extractor, decoder=decoder, config=config, name=name
         )
-        urls_extractor = self._create_component_from_model(
-            model=model.urls_extractor, decoder=decoder, config=config, name=name
+        download_target_extractor = self._create_component_from_model(
+            model=model.download_target_extractor, decoder=decoder, config=config, name=name
         )
         job_repository: AsyncJobRepository = AsyncHttpJobRepository(
             creation_requester=creation_requester,
@@ -2769,7 +2769,7 @@ def _get_download_retriever() -> SimpleRetrieverTestReadDecorator | SimpleRetrie
             delete_requester=delete_requester,
             status_extractor=status_extractor,
             status_mapping=self._create_async_job_status_mapping(model.status_mapping, config),
-            urls_extractor=urls_extractor,
+            download_target_extractor=download_target_extractor,
         )
 
         async_job_partition_router = AsyncJobPartitionRouter(

airbyte_cdk/sources/declarative/requesters/README.md

@@ -1,7 +1,7 @@
 # AsyncHttpJobRepository sequence diagram
 
 - Components marked as optional are not required and can be ignored.
-- if `url_requester` is not provided, `urls_extractor` will get urls from the `polling_job_response`
+- if `url_requester` is not provided, `download_target_extractor` will get urls from the `polling_job_response`
 - interpolation_context, e.g. `create_job_response` or `polling_job_response` can be obtained from stream_slice
 
 ```mermaid

airbyte_cdk/sources/declarative/requesters/http_job_repository.py

@@ -43,7 +43,7 @@ class AsyncHttpJobRepository(AsyncJobRepository):
     delete_requester: Optional[Requester]
     status_extractor: DpathExtractor
     status_mapping: Mapping[str, AsyncJobStatus]
-    urls_extractor: DpathExtractor
+    download_target_extractor: DpathExtractor
 
     job_timeout: Optional[timedelta] = None
     record_extractor: RecordExtractor = field(
@@ -216,7 +216,10 @@ def fetch_records(self, job: AsyncJob) -> Iterable[Mapping[str, Any]]:
             stream_slice = StreamSlice(
                 partition=job_slice.partition,
                 cursor_slice=job_slice.cursor_slice,
-                extra_fields={**job_slice.extra_fields, "url": url},
+                extra_fields={
+                    **job_slice.extra_fields,
+                    "download_target": url,
+                },
             )
             for message in self.download_retriever.read_records({}, stream_slice):
                 if isinstance(message, Record):
@@ -269,9 +272,11 @@ def _clean_up_job(self, job_id: str) -> None:
         del self._polling_job_response_by_id[job_id]
 
     def _get_create_job_stream_slice(self, job: AsyncJob) -> StreamSlice:
+        creation_response = self._create_job_response_by_id[job.api_job_id()].json()
         stream_slice = StreamSlice(
-            partition={"create_job_response": self._create_job_response_by_id[job.api_job_id()]},
+            partition={},
             cursor_slice={},
+            extra_fields={"creation_response": creation_response},
         )
         return stream_slice
 
@@ -292,4 +297,4 @@ def _get_download_url(self, job: AsyncJob) -> Iterable[str]:
                     failure_type=FailureType.system_error,
                 )
 
-        yield from self.urls_extractor.extract_records(url_response)  # type: ignore # we expect urls_extractor to always return list of strings
+        yield from self.download_target_extractor.extract_records(url_response)  # type: ignore # we expect download_target_extractor to always return list of strings

airbyte_cdk/sources/declarative/requesters/http_requester.py

@@ -85,7 +85,7 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self._parameters = parameters
 
         if self.error_handler is not None and hasattr(self.error_handler, "backoff_strategies"):
-            backoff_strategies = self.error_handler.backoff_strategies
+            backoff_strategies = self.error_handler.backoff_strategies  # type: ignore
         else:
             backoff_strategies = None
 
@@ -125,6 +125,12 @@ def get_path(
         kwargs = {
             "stream_slice": stream_slice,
             "next_page_token": next_page_token,
+            # update the interpolation context with extra fields, if passed.
+            **(
+                stream_slice.extra_fields
+                if stream_slice is not None and hasattr(stream_slice, "extra_fields")
+                else {}
+            ),
         }
         path = str(self._path.eval(self.config, **kwargs))
         return path.lstrip("/")

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -3607,7 +3607,7 @@ def test_create_async_retriever():
             "timeout": ["timeout"],
             "completed": ["ready"],
         },
-        "urls_extractor": {"type": "DpathExtractor", "field_path": ["urls"]},
+        "download_target_extractor": {"type": "DpathExtractor", "field_path": ["urls"]},
         "record_selector": {
             "type": "RecordSelector",
             "extractor": {"type": "DpathExtractor", "field_path": ["data"]},
@@ -3681,7 +3681,7 @@ def test_create_async_retriever():
     assert job_repository.abort_requester
     assert job_repository.delete_requester
     assert job_repository.status_extractor
-    assert job_repository.urls_extractor
+    assert job_repository.download_target_extractor
 
     selector = component.record_selector
     extractor = selector.extractor

unit_tests/sources/declarative/requesters/test_http_job_repository.py

@@ -69,7 +69,7 @@ def setUp(self) -> None:
         self._polling_job_requester = HttpRequester(
             name="stream <name>: polling",
             url_base=_URL_BASE,
-            path=_EXPORT_PATH + "/{{stream_slice['create_job_response'].json()['id']}}",
+            path=_EXPORT_PATH + "/{{creation_response['id']}}",
             error_handler=error_handler,
             http_method=HttpMethod.GET,
             config=_ANY_CONFIG,
@@ -84,7 +84,7 @@ def setUp(self) -> None:
             requester=HttpRequester(
                 name="stream <name>: fetch_result",
                 url_base="",
-                path="{{stream_slice.extra_fields['url']}}",
+                path="{{download_target}}",
                 error_handler=error_handler,
                 http_method=HttpMethod.GET,
                 config=_ANY_CONFIG,
@@ -143,7 +143,7 @@ def setUp(self) -> None:
                 "failure": AsyncJobStatus.FAILED,
                 "pending": AsyncJobStatus.RUNNING,
             },
-            urls_extractor=DpathExtractor(
+            download_target_extractor=DpathExtractor(
                 decoder=JsonDecoder(parameters={}),
                 field_path=["urls"],
                 config={},

unit_tests/sources/declarative/test_concurrent_declarative_source.py

@@ -299,7 +299,7 @@
                     "timeout": ["timeout"],
                     "completed": ["ready"],
                 },
-                "urls_extractor": {"type": "DpathExtractor", "field_path": ["urls"]},
+                "download_target_extractor": {"type": "DpathExtractor", "field_path": ["urls"]},
                 "record_selector": {
                     "type": "RecordSelector",
                     "extractor": {"type": "DpathExtractor", "field_path": []},