Merge branch 'main' into maxi297/filebased-fix-column-named-type

Author: maxi297

.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/declarative_component_schema.yaml

@@ -42,6 +42,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.
@@ -109,7 +113,7 @@ definitions:
         description: List of transformations (path and corresponding value) that will be added to the record.
         type: array
         items:
-          - "$ref": "#/definitions/AddedFieldDefinition"
+          "$ref": "#/definitions/AddedFieldDefinition"
       $parameters:
         type: object
         additionalProperties: true
@@ -1657,7 +1661,7 @@ definitions:
         description: List of potentially nested fields describing the full path of the field to extract. Use "*" to extract all values from an array. See more info in the [docs](https://docs.airbyte.com/connector-development/config-based/understanding-the-yaml-file/record-selector).
         type: array
         items:
-          - type: string
+          type: string
         interpolation_context:
           - config
         examples:
@@ -2073,29 +2077,29 @@ definitions:
         type: array
         default: []
         items:
-          - type: string
+          type: string
         interpolation_context:
           - config
       key_pointer:
         title: Key Path
         description: List of potentially nested fields describing the full path of the field key to extract.
         type: array
         items:
-          - type: string
+          type: string
         interpolation_context:
           - config
       type_pointer:
         title: Type Path
         description: List of potentially nested fields describing the full path of the field type to extract.
         type: array
         items:
-          - type: string
+          type: string
         interpolation_context:
           - config
       types_mapping:
         type: array
         items:
-          - "$ref": "#/definitions/TypesMap"
+          "$ref": "#/definitions/TypesMap"
       $parameters:
         type: object
         additionalProperties: true
@@ -2251,7 +2255,7 @@ definitions:
         description: A path to field that needs to be flattened.
         type: array
         items:
-          - type: string
+          type: string
         examples:
           - ["data"]
           - ["data", "*", "field"]
@@ -3526,7 +3530,7 @@ definitions:
         description: A list of potentially nested fields indicating the full path where value will be added or updated.
         type: array
         items:
-          - type: string
+          type: string
         interpolation_context:
           - config
           - components_values
@@ -3602,7 +3606,7 @@ definitions:
         description: A list of potentially nested fields indicating the full path in source config file where streams configs located.
         type: array
         items:
-          - type: string
+          type: string
         interpolation_context:
           - parameters
         examples:

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -93,7 +93,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 = (

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1871,6 +1871,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.",
@@ -1898,6 +1903,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.",

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -503,7 +503,7 @@
     IncrementingCountStreamStateConverter,
 )
 from airbyte_cdk.sources.streams.http.error_handlers.response_models import ResponseAction
-from airbyte_cdk.sources.types import Config
+from airbyte_cdk.sources.types import Config, ConnectionDefinition
 from airbyte_cdk.sources.utils.transform import TransformConfig, TypeTransformer
 
 ComponentDefinition = Mapping[str, Any]
@@ -527,6 +527,7 @@ def __init__(
         disable_resumable_full_refresh: bool = False,
         message_repository: Optional[MessageRepository] = None,
         connector_state_manager: Optional[ConnectorStateManager] = None,
+        max_concurrent_async_job_count: Optional[int] = None,
     ):
         self._init_mappings()
         self._limit_pages_fetched_per_slice = limit_pages_fetched_per_slice
@@ -540,6 +541,7 @@ def __init__(
         )
         self._connector_state_manager = connector_state_manager or ConnectorStateManager()
         self._api_budget: Optional[Union[APIBudget, HttpAPIBudget]] = None
+        self._job_tracker: JobTracker = JobTracker(max_concurrent_async_job_count or 1)
 
     def _init_mappings(self) -> None:
         self.PYDANTIC_MODEL_TO_CONSTRUCTOR: Mapping[Type[BaseModel], Callable[..., Any]] = {
@@ -2928,8 +2930,7 @@ def _get_download_retriever() -> SimpleRetrieverTestReadDecorator | SimpleRetrie
             job_orchestrator_factory=lambda stream_slices: AsyncJobOrchestrator(
                 job_repository,
                 stream_slices,
-                JobTracker(1),
-                # FIXME eventually make the number of concurrent jobs in the API configurable. Until then, we limit to 1
+                self._job_tracker,
                 self._message_repository,
                 has_bulk_parent=False,
                 # FIXME work would need to be done here in order to detect if a stream as a parent stream that is bulk

airbyte_cdk/sources/file_based/file_based_source.py

@@ -48,6 +48,9 @@
     FileBasedErrorsCollector,
     FileBasedSourceError,
 )
+from airbyte_cdk.sources.file_based.file_based_stream_permissions_reader import (
+    AbstractFileBasedStreamPermissionsReader,
+)
 from airbyte_cdk.sources.file_based.file_based_stream_reader import AbstractFileBasedStreamReader
 from airbyte_cdk.sources.file_based.file_types import default_parsers
 from airbyte_cdk.sources.file_based.file_types.file_type_parser import FileTypeParser
@@ -100,8 +103,10 @@ def __init__(
         cursor_cls: Type[
             Union[AbstractConcurrentFileBasedCursor, AbstractFileBasedCursor]
         ] = FileBasedConcurrentCursor,
+        stream_permissions_reader: Optional[AbstractFileBasedStreamPermissionsReader] = None,
     ):
         self.stream_reader = stream_reader
+        self.stream_permissions_reader = stream_permissions_reader
         self.spec_class = spec_class
         self.config = config
         self.catalog = catalog
@@ -234,6 +239,8 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         try:
             parsed_config = self._get_parsed_config(config)
             self.stream_reader.config = parsed_config
+            if self.stream_permissions_reader:
+                self.stream_permissions_reader.config = parsed_config
             streams: List[Stream] = []
             for stream_config in parsed_config.streams:
                 # Like state_manager, `catalog_stream` may be None during `check`
@@ -337,9 +344,23 @@ def _make_default_stream(
             preserve_directory_structure=preserve_directory_structure(parsed_config),
         )
 
+    def _ensure_permissions_reader_available(self) -> None:
+        """
+        Validates that a stream permissions reader is available.
+        Raises a ValueError if the reader is not provided.
+        """
+        if not self.stream_permissions_reader:
+            raise ValueError(
+                "Stream permissions reader is required for streams that use permissions transfer mode."
+            )
+
     def _make_permissions_stream(
         self, stream_config: FileBasedStreamConfig, cursor: Optional[AbstractFileBasedCursor]
     ) -> AbstractFileBasedStream:
+        """
+        Creates a stream that reads permissions from files.
+        """
+        self._ensure_permissions_reader_available()
         return PermissionsFileBasedStream(
             config=stream_config,
             catalog_schema=self.stream_schemas.get(stream_config.name),
@@ -350,6 +371,7 @@ def _make_permissions_stream(
             validation_policy=self._validate_and_get_validation_policy(stream_config),
             errors_collector=self.errors_collector,
             cursor=cursor,
+            stream_permissions_reader=self.stream_permissions_reader,  # type: ignore
         )
 
     def _make_file_based_stream(
@@ -370,9 +392,10 @@ def _make_file_based_stream(
     def _make_identities_stream(
         self,
     ) -> Stream:
+        self._ensure_permissions_reader_available()
         return FileIdentitiesStream(
             catalog_schema=self.stream_schemas.get(FileIdentitiesStream.IDENTITIES_STREAM_NAME),
-            stream_reader=self.stream_reader,
+            stream_permissions_reader=self.stream_permissions_reader,  # type: ignore
             discovery_policy=self.discovery_policy,
             errors_collector=self.errors_collector,
         )