Merge branch 'main' into lazebnyi/add-condition-to-add-fields

Author: lazebnyi

.github/workflows/python_dependency_analysis.yml

@@ -0,0 +1,38 @@
+name: Dependency Analysis
+
+on:
+  push:
+    branches:
+      - main
+    paths:
+      - "airbyte_cdk/**"
+      - "poetry.lock"
+      - "pyproject.toml"
+  pull_request:
+    paths:
+      - "airbyte_cdk/**"
+      - "poetry.lock"
+      - "pyproject.toml"
+
+jobs:
+  dependency-analysis:
+    name: Dependency Analysis with Deptry
+    runs-on: ubuntu-24.04
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v4
+    - name: Set up Python
+      uses: actions/setup-python@v5
+      with:
+        python-version: '3.10'
+    - name: Set up Poetry
+      uses: Gr1N/setup-poetry@v9
+      with:
+        poetry-version: "2.0.1"
+    - name: Install dependencies
+      run: poetry install --all-extras
+
+    # Job-specific step(s):
+    - name: Run Deptry
+      run: |
+        poetry run deptry .

airbyte_cdk/sources/declarative/async_job/job_tracker.py

@@ -17,7 +17,11 @@ class ConcurrentJobLimitReached(Exception):
 class JobTracker:
     def __init__(self, limit: int):
         self._jobs: Set[str] = set()
-        self._limit = limit
+        if limit < 1:
+            LOGGER.warning(
+                f"The `max_concurrent_async_job_count` property is less than 1: {limit}. Setting to 1. Please update the source manifest to set a valid value."
+            )
+        self._limit = 1 if limit < 1 else limit
         self._lock = threading.Lock()
 
     def try_to_get_intent(self) -> str:

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -162,6 +162,10 @@ def read(
         else:
             filtered_catalog = catalog
 
+        # It is no need run read for synchronous streams if they are not exists.
+        if not filtered_catalog.streams:
+            return
+
         yield from super().read(logger, config, filtered_catalog, state)
 
     def discover(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteCatalog:
@@ -201,6 +205,18 @@ def _group_streams(
             # Some low-code sources use a combination of DeclarativeStream and regular Python streams. We can't inspect
             # these legacy Python streams the way we do low-code streams to determine if they are concurrent compatible,
             # so we need to treat them as synchronous
+
+            if name_to_stream_mapping[declarative_stream.name]["type"] == "StateDelegatingStream":
+                stream_state = self._connector_state_manager.get_stream_state(
+                    stream_name=declarative_stream.name, namespace=declarative_stream.namespace
+                )
+
+                name_to_stream_mapping[declarative_stream.name] = (
+                    name_to_stream_mapping[declarative_stream.name]["incremental_stream"]
+                    if stream_state
+                    else name_to_stream_mapping[declarative_stream.name]["full_refresh_stream"]
+                )
+
             if isinstance(declarative_stream, DeclarativeStream) and (
                 name_to_stream_mapping[declarative_stream.name]["retriever"]["type"]
                 == "SimpleRetriever"

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -24,7 +24,9 @@ properties:
   streams:
     type: array
     items:
-      "$ref": "#/definitions/DeclarativeStream"
+      anyOf:
+        - "$ref": "#/definitions/DeclarativeStream"
+        - "$ref": "#/definitions/StateDelegatingStream"
   dynamic_streams:
     type: array
     items:
@@ -42,6 +44,10 @@ properties:
     "$ref": "#/definitions/ConcurrencyLevel"
   api_budget:
     "$ref": "#/definitions/HTTPAPIBudget"
+  max_concurrent_async_job_count:
+    title: Maximum Concurrent Asynchronous Jobs
+    description: Maximum number of concurrent asynchronous jobs to run. This property is only relevant for sources/streams that support asynchronous job execution through the AsyncRetriever (e.g. a report-based stream that initiates a job, polls the job status, and then fetches the job results). This is often set by the API's maximum number of concurrent jobs on the account level. Refer to the API's documentation for this information.
+    type: integer
   metadata:
     type: object
     description: For internal Airbyte use only - DO NOT modify manually. Used by consumers of declarative manifests for storing related metadata.
@@ -2272,6 +2278,10 @@ definitions:
         title: Delete Origin Value
         description: Whether to delete the origin value or keep it. Default is False.
         type: boolean
+      replace_record:
+        title: Replace Origin Record
+        description: Whether to replace the origin record or not. Default is False.
+        type: boolean
       $parameters:
         type: object
         additionalProperties: true
@@ -2890,7 +2900,9 @@ definitions:
       stream:
         title: Parent Stream
         description: Reference to the parent stream.
-        "$ref": "#/definitions/DeclarativeStream"
+        anyOf:
+          - "$ref": "#/definitions/DeclarativeStream"
+          - "$ref": "#/definitions/StateDelegatingStream"
       partition_field:
         title: Current Parent Key Value Identifier
         description: While iterating over parent records during a sync, the parent_key value can be referenced by using this field.
@@ -3163,6 +3175,36 @@ definitions:
       $parameters:
         type: object
         additionalProperties: true
+  StateDelegatingStream:
+    description: (This component is experimental. Use at your own risk.) Orchestrate the retriever's usage based on the state value.
+    type: object
+    required:
+      - type
+      - name
+      - full_refresh_stream
+      - incremental_stream
+    properties:
+      type:
+        type: string
+        enum: [ StateDelegatingStream ]
+      name:
+        title: Name
+        description: The stream name.
+        type: string
+        default: ""
+        example:
+          - "Users"
+      full_refresh_stream:
+        title: Retriever
+        description: Component used to coordinate how records are extracted across stream slices and request pages when the state is empty or not provided.
+        "$ref": "#/definitions/DeclarativeStream"
+      incremental_stream:
+        title: Retriever
+        description: Component used to coordinate how records are extracted across stream slices and request pages when the state provided.
+        "$ref": "#/definitions/DeclarativeStream"
+      $parameters:
+        type: object
+        additionalProperties: true
   SimpleRetriever:
     description: Retrieves records by synchronously sending requests to fetch records. The retriever acts as an orchestrator between the requester, the record selector, the paginator, and the partition router.
     type: object

airbyte_cdk/sources/declarative/declarative_stream.py

@@ -14,6 +14,7 @@
 from airbyte_cdk.sources.declarative.interpolation import InterpolatedString
 from airbyte_cdk.sources.declarative.migrations.state_migration import StateMigration
 from airbyte_cdk.sources.declarative.retrievers import SimpleRetriever
+from airbyte_cdk.sources.declarative.retrievers.async_retriever import AsyncRetriever
 from airbyte_cdk.sources.declarative.retrievers.retriever import Retriever
 from airbyte_cdk.sources.declarative.schema import DefaultSchemaLoader
 from airbyte_cdk.sources.declarative.schema.schema_loader import SchemaLoader
@@ -76,11 +77,17 @@ def primary_key(self, value: str) -> None:
 
     @property
     def exit_on_rate_limit(self) -> bool:
+        if isinstance(self.retriever, AsyncRetriever):
+            return self.retriever.exit_on_rate_limit
+
         return self.retriever.requester.exit_on_rate_limit  # type: ignore # abstract Retriever class has not requester attribute
 
     @exit_on_rate_limit.setter
     def exit_on_rate_limit(self, value: bool) -> None:
-        self.retriever.requester.exit_on_rate_limit = value  # type: ignore[attr-defined]
+        if isinstance(self.retriever, AsyncRetriever):
+            self.retriever.exit_on_rate_limit = value
+        else:
+            self.retriever.requester.exit_on_rate_limit = value  # type: ignore[attr-defined]
 
     @property  # type: ignore
     def name(self) -> str:

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -30,6 +30,9 @@
     DeclarativeStream as DeclarativeStreamModel,
 )
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import Spec as SpecModel
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    StateDelegatingStream as StateDelegatingStreamModel,
+)
 from airbyte_cdk.sources.declarative.parsers.custom_code_compiler import (
     get_registered_components_module,
 )
@@ -93,7 +96,10 @@ def __init__(
         self._constructor = (
             component_factory
             if component_factory
-            else ModelToComponentFactory(emit_connector_builder_messages)
+            else ModelToComponentFactory(
+                emit_connector_builder_messages,
+                max_concurrent_async_job_count=source_config.get("max_concurrent_async_job_count"),
+            )
         )
         self._message_repository = self._constructor.get_message_repository()
         self._slice_logger: SliceLogger = (
@@ -143,7 +149,9 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
 
         source_streams = [
             self._constructor.create_component(
-                DeclarativeStreamModel,
+                StateDelegatingStreamModel
+                if stream_config.get("type") == StateDelegatingStreamModel.__name__
+                else DeclarativeStreamModel,
                 stream_config,
                 config,
                 emit_connector_builder_messages=self._emit_connector_builder_messages,
@@ -162,7 +170,15 @@ def _initialize_cache_for_parent_streams(
         def update_with_cache_parent_configs(parent_configs: list[dict[str, Any]]) -> None:
             for parent_config in parent_configs:
                 parent_streams.add(parent_config["stream"]["name"])
-                parent_config["stream"]["retriever"]["requester"]["use_cache"] = True
+                if parent_config["stream"]["type"] == "StateDelegatingStream":
+                    parent_config["stream"]["full_refresh_stream"]["retriever"]["requester"][
+                        "use_cache"
+                    ] = True
+                    parent_config["stream"]["incremental_stream"]["retriever"]["requester"][
+                        "use_cache"
+                    ] = True
+                else:
+                    parent_config["stream"]["retriever"]["requester"]["use_cache"] = True
 
         for stream_config in stream_configs:
             if stream_config.get("incremental_sync", {}).get("parent_stream"):
@@ -185,7 +201,15 @@ def update_with_cache_parent_configs(parent_configs: list[dict[str, Any]]) -> No
 
         for stream_config in stream_configs:
             if stream_config["name"] in parent_streams:
-                stream_config["retriever"]["requester"]["use_cache"] = True
+                if stream_config["type"] == "StateDelegatingStream":
+                    stream_config["full_refresh_stream"]["retriever"]["requester"]["use_cache"] = (
+                        True
+                    )
+                    stream_config["incremental_stream"]["retriever"]["requester"]["use_cache"] = (
+                        True
+                    )
+                else:
+                    stream_config["retriever"]["requester"]["use_cache"] = True
 
         return stream_configs
 

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -877,6 +877,11 @@ class DpathFlattenFields(BaseModel):
         description="Whether to delete the origin value or keep it. Default is False.",
         title="Delete Origin Value",
     )
+    replace_record: Optional[bool] = Field(
+        None,
+        description="Whether to replace the origin record or not. Default is False.",
+        title="Replace Origin Record",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
@@ -1870,7 +1875,7 @@ class Config:
 
     type: Literal["DeclarativeSource"]
     check: Union[CheckStream, CheckDynamicStream]
-    streams: List[DeclarativeStream]
+    streams: List[Union[DeclarativeStream, StateDelegatingStream]]
     dynamic_streams: Optional[List[DynamicDeclarativeStream]] = None
     version: str = Field(
         ...,
@@ -1881,6 +1886,11 @@ class Config:
     spec: Optional[Spec] = None
     concurrency_level: Optional[ConcurrencyLevel] = None
     api_budget: Optional[HTTPAPIBudget] = None
+    max_concurrent_async_job_count: Optional[int] = Field(
+        None,
+        description="Maximum number of concurrent asynchronous jobs to run. This property is only relevant for sources/streams that support asynchronous job execution through the AsyncRetriever (e.g. a report-based stream that initiates a job, polls the job status, and then fetches the job results). This is often set by the API's maximum number of concurrent jobs on the account level. Refer to the API's documentation for this information.",
+        title="Maximum Concurrent Asynchronous Jobs",
+    )
     metadata: Optional[Dict[str, Any]] = Field(
         None,
         description="For internal Airbyte use only - DO NOT modify manually. Used by consumers of declarative manifests for storing related metadata.",
@@ -1897,7 +1907,7 @@ class Config:
 
     type: Literal["DeclarativeSource"]
     check: Union[CheckStream, CheckDynamicStream]
-    streams: Optional[List[DeclarativeStream]] = None
+    streams: Optional[List[Union[DeclarativeStream, StateDelegatingStream]]] = None
     dynamic_streams: List[DynamicDeclarativeStream]
     version: str = Field(
         ...,
@@ -1908,6 +1918,11 @@ class Config:
     spec: Optional[Spec] = None
     concurrency_level: Optional[ConcurrencyLevel] = None
     api_budget: Optional[HTTPAPIBudget] = None
+    max_concurrent_async_job_count: Optional[int] = Field(
+        None,
+        description="Maximum number of concurrent asynchronous jobs to run. This property is only relevant for sources/streams that support asynchronous job execution through the AsyncRetriever (e.g. a report-based stream that initiates a job, polls the job status, and then fetches the job results). This is often set by the API's maximum number of concurrent jobs on the account level. Refer to the API's documentation for this information.",
+        title="Maximum Concurrent Asynchronous Jobs",
+    )
     metadata: Optional[Dict[str, Any]] = Field(
         None,
         description="For internal Airbyte use only - DO NOT modify manually. Used by consumers of declarative manifests for storing related metadata.",
@@ -2211,7 +2226,7 @@ class ParentStreamConfig(BaseModel):
         examples=["id", "{{ config['parent_record_id'] }}"],
         title="Parent Key",
     )
-    stream: DeclarativeStream = Field(
+    stream: Union[DeclarativeStream, StateDelegatingStream] = Field(
         ..., description="Reference to the parent stream.", title="Parent Stream"
     )
     partition_field: str = Field(
@@ -2238,6 +2253,22 @@ class ParentStreamConfig(BaseModel):
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
+class StateDelegatingStream(BaseModel):
+    type: Literal["StateDelegatingStream"]
+    name: str = Field(..., description="The stream name.", example=["Users"], title="Name")
+    full_refresh_stream: DeclarativeStream = Field(
+        ...,
+        description="Component used to coordinate how records are extracted across stream slices and request pages when the state is empty or not provided.",
+        title="Retriever",
+    )
+    incremental_stream: DeclarativeStream = Field(
+        ...,
+        description="Component used to coordinate how records are extracted across stream slices and request pages when the state provided.",
+        title="Retriever",
+    )
+    parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
+
+
 class SimpleRetriever(BaseModel):
     type: Literal["SimpleRetriever"]
     record_selector: RecordSelector = Field(
@@ -2423,5 +2454,6 @@ class DynamicDeclarativeStream(BaseModel):
 DeclarativeStream.update_forward_refs()
 SessionTokenAuthenticator.update_forward_refs()
 DynamicSchemaLoader.update_forward_refs()
+ParentStreamConfig.update_forward_refs()
 SimpleRetriever.update_forward_refs()
 AsyncRetriever.update_forward_refs()