fix: Async Retriever change url path for download retriever

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -2254,7 +2254,7 @@ def create_async_retriever(
                 extractor=download_extractor,
                 name=name,
                 record_filter=None,
-                transformations=[],
+                transformations=transformations,
                 schema_normalization=TypeTransformer(TransformConfig.NoTransform),
                 config=config,
                 parameters={},

airbyte_cdk/sources/declarative/partition_routers/substream_partition_router.py

@@ -4,7 +4,7 @@
 import copy
 import logging
 from dataclasses import InitVar, dataclass
-from typing import TYPE_CHECKING, Any, Iterable, List, Mapping, Optional, Union
+from typing import TYPE_CHECKING, Any, Iterable, List, Mapping, MutableMapping, Optional, Union
 
 import dpath
 
@@ -215,7 +215,7 @@ def _extract_extra_fields(
         self,
         parent_record: Mapping[str, Any] | AirbyteMessage,
         extra_fields: Optional[List[List[str]]] = None,
-    ) -> Mapping[str, Any]:
+    ) -> MutableMapping[str, Any]:
         """
         Extracts additional fields specified by their paths from the parent record.
 

airbyte_cdk/sources/declarative/requesters/http_job_repository.py

@@ -189,7 +189,8 @@ def fetch_records(self, job: AsyncJob) -> Iterable[Mapping[str, Any]]:
         for url in self.urls_extractor.extract_records(
             self._polling_job_response_by_id[job.api_job_id()]
         ):
-            stream_slice: StreamSlice = StreamSlice(partition={"url": url}, cursor_slice={})
+            stream_slice = job.job_parameters()
+            stream_slice.extra_fields.update({"url": url})
             for message in self.download_retriever.read_records({}, stream_slice):
                 if isinstance(message, Record):
                     yield message.data

airbyte_cdk/sources/types.py

@@ -4,7 +4,17 @@
 
 from __future__ import annotations
 
-from typing import Any, ItemsView, Iterator, KeysView, List, Mapping, Optional, ValuesView
+from typing import (
+    Any,
+    ItemsView,
+    Iterator,
+    KeysView,
+    List,
+    Mapping,
+    MutableMapping,
+    Optional,
+    ValuesView,
+)
 
 import orjson
 
@@ -68,7 +78,7 @@ def __init__(
         *,
         partition: Mapping[str, Any],
         cursor_slice: Mapping[str, Any],
-        extra_fields: Optional[Mapping[str, Any]] = None,
+        extra_fields: Optional[MutableMapping[str, Any]] = None,
     ) -> None:
         """
         :param partition: The partition keys representing a unique partition in the stream.
@@ -102,7 +112,7 @@ def cursor_slice(self) -> Mapping[str, Any]:
         return c
 
     @property
-    def extra_fields(self) -> Mapping[str, Any]:
+    def extra_fields(self) -> MutableMapping[str, Any]:
         """Returns the extra fields that are not part of the partition."""
         return self._extra_fields
 
@@ -152,3 +162,6 @@ def __json_serializable__(self) -> Any:
 
     def __hash__(self) -> int:
         return hash(orjson.dumps(self._stream_slice, option=orjson.OPT_SORT_KEYS))
+
+    def __bool__(self) -> bool:
+        return bool(self._stream_slice) or bool(self._extra_fields)

unit_tests/sources/declarative/requesters/test_http_job_repository.py

@@ -84,7 +84,7 @@ def setUp(self) -> None:
             requester=HttpRequester(
                 name="stream <name>: fetch_result",
                 url_base="",
-                path="{{stream_slice['url']}}",
+                path="{{stream_slice.extra_fields['url']}}",
                 error_handler=error_handler,
                 http_method=HttpMethod.GET,
                 config=_ANY_CONFIG,