Merge branch 'main' into requests-env-settings

Author: aaronsteers

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -20,7 +20,7 @@
 
 import orjson
 import yaml
-from airbyte_protocol_dataclasses.models import Level
+from airbyte_protocol_dataclasses.models import AirbyteStreamStatus, Level, StreamDescriptor
 from jsonschema.exceptions import ValidationError
 from jsonschema.validators import validate
 
@@ -88,6 +88,7 @@
     DebugSliceLogger,
     SliceLogger,
 )
+from airbyte_cdk.utils.stream_status_utils import as_airbyte_message
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 
 
@@ -630,15 +631,22 @@ def _dynamic_stream_configs(
 
         return dynamic_stream_configs
 
-    @staticmethod
     def _select_streams(
-        streams: List[AbstractStream], configured_catalog: ConfiguredAirbyteCatalog
+        self, streams: List[AbstractStream], configured_catalog: ConfiguredAirbyteCatalog
     ) -> List[AbstractStream]:
         stream_name_to_instance: Mapping[str, AbstractStream] = {s.name: s for s in streams}
         abstract_streams: List[AbstractStream] = []
         for configured_stream in configured_catalog.streams:
             stream_instance = stream_name_to_instance.get(configured_stream.stream.name)
             if stream_instance:
                 abstract_streams.append(stream_instance)
-
+            else:
+                # Previous behavior in the legacy synchronous CDK was to also raise an error TRACE message if
+                # the source was configured with raise_exception_on_missing_stream=True. This was used on very
+                # few sources like facebook-marketing and google-ads. We decided not to port this feature over,
+                # but we can do so if we feel it necessary. With the current behavior,we should still result
+                # in a partial failure since missing streams will be marked as INCOMPLETE.
+                self._message_repository.emit_message(
+                    as_airbyte_message(configured_stream.stream, AirbyteStreamStatus.INCOMPLETE)
+                )
         return abstract_streams

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -3797,7 +3797,6 @@ definitions:
       - polling_requester
       - download_requester
       - status_extractor
-      - download_target_extractor
     properties:
       type:
         type: string

airbyte_cdk/sources/declarative/incremental/concurrent_partition_cursor.py

@@ -189,6 +189,7 @@ def __init__(
         # FIXME this is a temporary field the time of the migration from declarative cursors to concurrent ones
         self._attempt_to_create_cursor_if_not_provided = attempt_to_create_cursor_if_not_provided
         self._synced_some_data = False
+        self._logged_regarding_datetime_format_error = False
 
     @property
     def cursor_field(self) -> CursorField:
@@ -518,10 +519,23 @@ def observe(self, record: Record) -> None:
         except ValueError:
             return
 
+        try:
+            record_cursor = self._connector_state_converter.output_format(
+                self._connector_state_converter.parse_value(record_cursor_value)
+            )
+        except ValueError as exception:
+            if not self._logged_regarding_datetime_format_error:
+                logger.warning(
+                    "Skipping cursor update for stream '%s': failed to parse cursor field '%s' value %r: %s",
+                    self._stream_name,
+                    self._cursor_field.cursor_field_key,
+                    record_cursor_value,
+                    exception,
+                )
+                self._logged_regarding_datetime_format_error = True
+            return
+
         self._synced_some_data = True
-        record_cursor = self._connector_state_converter.output_format(
-            self._connector_state_converter.parse_value(record_cursor_value)
-        )
         self._update_global_cursor(record_cursor)
         if not self._use_global_cursor:
             self._cursor_per_partition[

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

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

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -1976,7 +1976,10 @@ def create_default_stream(
         primary_key = model.primary_key.__root__ if model.primary_key else None
 
         partition_router = self._build_stream_slicer_from_partition_router(
-            model.retriever, config, stream_name=model.name
+            model.retriever,
+            config,
+            stream_name=model.name,
+            **kwargs,
         )
         concurrent_cursor = self._build_concurrent_cursor(model, partition_router, config)
         if model.incremental_sync and isinstance(model.incremental_sync, DatetimeBasedCursorModel):
@@ -2155,10 +2158,11 @@ def _build_stream_slicer_from_partition_router(
         ],
         config: Config,
         stream_name: Optional[str] = None,
+        **kwargs: Any,
     ) -> PartitionRouter:
         if (
             hasattr(model, "partition_router")
-            and isinstance(model, SimpleRetrieverModel | AsyncRetrieverModel)
+            and isinstance(model, (SimpleRetrieverModel, AsyncRetrieverModel, CustomRetrieverModel))
             and model.partition_router
         ):
             stream_slicer_model = model.partition_router
@@ -2172,6 +2176,23 @@ def _build_stream_slicer_from_partition_router(
                     ],
                     parameters={},
                 )
+            elif isinstance(stream_slicer_model, dict):
+                # partition router comes from CustomRetrieverModel therefore has not been parsed as a model
+                params = stream_slicer_model.get("$parameters")
+                if not isinstance(params, dict):
+                    params = {}
+                    stream_slicer_model["$parameters"] = params
+
+                if stream_name is not None:
+                    params["stream_name"] = stream_name
+
+                return self._create_nested_component(  # type: ignore[no-any-return] # There is no guarantee that this will return a stream slicer. If not, we expect an AttributeError during the call to `stream_slices`
+                    model,
+                    "partition_router",
+                    stream_slicer_model,
+                    config,
+                    **kwargs,
+                )
             else:
                 return self._create_component_from_model(  # type: ignore[no-any-return] # Will be created PartitionRouter as stream_slicer_model is model.partition_router
                     model=stream_slicer_model, config=config, stream_name=stream_name or ""
@@ -2886,7 +2907,7 @@ def create_page_increment(
         )
 
     def create_parent_stream_config(
-        self, model: ParentStreamConfigModel, config: Config, stream_name: str, **kwargs: Any
+        self, model: ParentStreamConfigModel, config: Config, *, stream_name: str, **kwargs: Any
     ) -> ParentStreamConfig:
         declarative_stream = self._create_component_from_model(
             model.stream,
@@ -3446,6 +3467,11 @@ def create_async_retriever(
         transformations: List[RecordTransformation],
         **kwargs: Any,
     ) -> AsyncRetriever:
+        if model.download_target_requester and not model.download_target_extractor:
+            raise ValueError(
+                f"`download_target_extractor` required if using a `download_target_requester`"
+            )
+
         def _get_download_retriever(
             requester: Requester, extractor: RecordExtractor, _decoder: Decoder
         ) -> SimpleRetriever:
@@ -3603,11 +3629,15 @@ def _get_job_timeout() -> datetime.timedelta:
         status_extractor = self._create_component_from_model(
             model=model.status_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,
+        download_target_extractor = (
+            self._create_component_from_model(
+                model=model.download_target_extractor,
+                decoder=decoder,
+                config=config,
+                name=name,
+            )
+            if model.download_target_extractor
+            else None
         )
 
         job_repository: AsyncJobRepository = AsyncHttpJobRepository(
@@ -3693,14 +3723,19 @@ def create_spec(self, model: SpecModel, config: Config, **kwargs: Any) -> Spec:
         )
 
     def create_substream_partition_router(
-        self, model: SubstreamPartitionRouterModel, config: Config, **kwargs: Any
+        self,
+        model: SubstreamPartitionRouterModel,
+        config: Config,
+        *,
+        stream_name: str,
+        **kwargs: Any,
     ) -> SubstreamPartitionRouter:
         parent_stream_configs = []
         if model.parent_stream_configs:
             parent_stream_configs.extend(
                 [
                     self.create_parent_stream_config_with_substream_wrapper(
-                        model=parent_stream_config, config=config, **kwargs
+                        model=parent_stream_config, config=config, stream_name=stream_name, **kwargs
                     )
                     for parent_stream_config in model.parent_stream_configs
                 ]
@@ -3720,7 +3755,7 @@ def create_parent_stream_config_with_substream_wrapper(
 
         # This flag will be used exclusively for StateDelegatingStream when a parent stream is created
         has_parent_state = bool(
-            self._connector_state_manager.get_stream_state(kwargs.get("stream_name", ""), None)
+            self._connector_state_manager.get_stream_state(stream_name, None)
             if model.incremental_dependency
             else False
         )
@@ -4113,11 +4148,17 @@ def set_api_budget(self, component_definition: ComponentDefinition, config: Conf
         )
 
     def create_grouping_partition_router(
-        self, model: GroupingPartitionRouterModel, config: Config, **kwargs: Any
+        self,
+        model: GroupingPartitionRouterModel,
+        config: Config,
+        *,
+        stream_name: str,
+        **kwargs: Any,
     ) -> GroupingPartitionRouter:
         underlying_router = self._create_component_from_model(
             model=model.underlying_partition_router,
             config=config,
+            stream_name=stream_name,
             **kwargs,
         )
         if model.group_size < 1:

airbyte_cdk/sources/declarative/requesters/README.md

@@ -1,8 +1,19 @@
+# Download Target and Download Requester
+
+- The `creation_response` and `polling_response` interpolation contexts are always available during the job download step of the process.
+
+- The`download_target` interpolation context is generated by the `download_target_extractor` and made available to the job download step as well.
+  - if `download_target_requester` is not provided, `download_target_extractor` will get urls from the `polling_response`
+  - if `download_target_requester` is provided, an additional request will be made to fetch job download targets and `download_target_extractor` will operate on that response
+
+## Some important considerations
+
+- **Note:** If the `download_target_extractor` and `download_target_requester` are not defined, a single job download request will be made without the `download_target` context.
+- **Note:** The `download_target_extractor` is required (not optional) if using a `download_target_requester`
+
 # AsyncHttpJobRepository sequence diagram
 
 - Components marked as optional are not required and can be ignored.
-- if `download_target_requester` is not provided, `download_target_extractor` will get urls from the `polling_response`
-- interpolation_context, e.g. `creation_response` or `polling_response` can be obtained from stream_slice
 
 ```mermaid
 ---
@@ -37,7 +48,7 @@ sequenceDiagram
         UrlRequester -->> AsyncHttpJobRepository: Download URLs
 
         AsyncHttpJobRepository ->> DownloadRetriever: Download reports
-        DownloadRetriever ->> Reporting Server: Retrieve report data (interpolation_context: `url`)
+        DownloadRetriever ->> Reporting Server: Retrieve report data (interpolation_context: `download_target`, `creation_response`, `polling_response`)
         Reporting Server -->> DownloadRetriever: Report data
         DownloadRetriever -->> AsyncHttpJobRepository: Report data
     else Status: Failed

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]
-    download_target_extractor: DpathExtractor
+    download_target_extractor: Optional[DpathExtractor]
 
     # timeout for the job to be completed, passed from `polling_job_timeout`
     job_timeout: Optional[timedelta] = None
@@ -213,14 +213,16 @@ def fetch_records(self, job: AsyncJob) -> Iterable[Mapping[str, Any]]:
 
         """
 
-        for target_url in self._get_download_targets(job):
+        for download_target in self._get_download_targets(job):
             job_slice = job.job_parameters()
             stream_slice = StreamSlice(
                 partition=job_slice.partition,
                 cursor_slice=job_slice.cursor_slice,
                 extra_fields={
                     **job_slice.extra_fields,
-                    "download_target": target_url,
+                    "download_target": download_target,
+                    "creation_response": self._get_creation_response_interpolation_context(job),
+                    "polling_response": self._get_polling_response_interpolation_context(job),
                 },
             )
             for message in self.download_retriever.read_records({}, stream_slice):
@@ -330,9 +332,27 @@ def _get_create_job_stream_slice(self, job: AsyncJob) -> StreamSlice:
         )
 
     def _get_download_targets(self, job: AsyncJob) -> Iterable[str]:
-        if not self.download_target_requester:
-            url_response = self._polling_job_response_by_id[job.api_job_id()]
-        else:
+        """Returns an iterable of strings to help target requests for downloading async jobs."""
+        # If neither download_target_extractor nor download_target_requester are provided, yield a single empty string
+        # to express the need to make a single download request without any download_target value
+        if not self.download_target_extractor:
+            if not self.download_target_requester:
+                lazy_log(
+                    LOGGER,
+                    logging.DEBUG,
+                    lambda: "No download_target_extractor or download_target_requester provided. Will attempt a single download request without a `download_target`.",
+                )
+                yield ""
+                return
+            else:
+                raise AirbyteTracedException(
+                    internal_message="Must define a `download_target_extractor` when using a `download_target_requester`.",
+                    failure_type=FailureType.config_error,
+                )
+
+        # We have a download_target_extractor, use it to extract the donload_target
+        if self.download_target_requester:
+            # if a download_target_requester if defined, we extract from the response of a request specifically for download targets.
             stream_slice: StreamSlice = StreamSlice(
                 partition={},
                 cursor_slice={},
@@ -346,5 +366,8 @@ def _get_download_targets(self, job: AsyncJob) -> Iterable[str]:
                     internal_message="Always expect a response or an exception from download_target_requester",
                     failure_type=FailureType.system_error,
                 )
+        else:
+            # if no download_target_requester is defined, we extract from the polling response
+            url_response = self._polling_job_response_by_id[job.api_job_id()]
 
         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/utils/transform.py

@@ -8,6 +8,7 @@
 from typing import TYPE_CHECKING, Any, Callable, Dict, Generator, Mapping, Optional, cast
 
 from jsonschema import Draft7Validator, ValidationError, validators
+from jsonschema.protocols import Validator
 from referencing import Registry, Resource
 from referencing._core import Resolver
 from referencing.exceptions import Unresolvable
@@ -17,12 +18,6 @@
 
 from .schema_helpers import get_ref_resolver_registry
 
-try:
-    from jsonschema.validators import Validator
-except:
-    from jsonschema import Validator
-
-
 MAX_NESTING_DEPTH = 3
 json_to_python_simple = {
     "string": str,

airbyte_cdk/utils/datetime_helpers.py

@@ -82,11 +82,10 @@
 """
 
 from datetime import datetime, timedelta, timezone
-from typing import Any, Optional, Union, overload
+from typing import Any, Union, overload
 
 from dateutil import parser
-from typing_extensions import Never
-from whenever import Instant, LocalDateTime, ZonedDateTime
+from whenever import Instant
 
 
 class AirbyteDateTime(datetime):

bin/generate-component-manifest-dagger.sh

@@ -7,5 +7,4 @@
 
 set -e
 
-pip install dagger-io==0.13.3
 python bin/generate_component_manifest_files.py