file-api: move catalog pass from streams() to ModelToComponentFactory.__init__() so we don't mess the interface signature

Author: aldogonzalez8

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -88,6 +88,7 @@ def __init__(
             emit_connector_builder_messages=emit_connector_builder_messages,
             disable_resumable_full_refresh=True,
             connector_state_manager=self._connector_state_manager,
+            catalog=catalog,
         )
 
         super().__init__(
@@ -139,7 +140,7 @@ def read(
         catalog: ConfiguredAirbyteCatalog,
         state: Optional[List[AirbyteStateMessage]] = None,
     ) -> Iterator[AirbyteMessage]:
-        concurrent_streams, _ = self._group_streams(config=config, catalog=catalog)
+        concurrent_streams, _ = self._group_streams(config=config)
 
         # ConcurrentReadProcessor pops streams that are finished being read so before syncing, the names of
         # the concurrent streams must be saved so that they can be removed from the catalog before starting
@@ -180,9 +181,7 @@ def discover(self, logger: logging.Logger, config: Mapping[str, Any]) -> Airbyte
             ]
         )
 
-    def streams(
-        self, config: Mapping[str, Any], catalog: ConfiguredAirbyteCatalog | None = None
-    ) -> List[Stream]:
+    def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         """
         The `streams` method is used as part of the AbstractSource in the following cases:
         * ConcurrentDeclarativeSource.check -> ManifestDeclarativeSource.check -> AbstractSource.check -> DeclarativeSource.check_connection -> CheckStream.check_connection -> streams
@@ -191,10 +190,10 @@ def streams(
 
         In both case, we will assume that calling the DeclarativeStream is perfectly fine as the result for these is the same regardless of if it is a DeclarativeStream or a DefaultStream (concurrent). This should simply be removed once we have moved away from the mentioned code paths above.
         """
-        return super().streams(config, catalog=catalog)
+        return super().streams(config)
 
     def _group_streams(
-        self, config: Mapping[str, Any], catalog: ConfiguredAirbyteCatalog | None = None
+        self, config: Mapping[str, Any]
     ) -> Tuple[List[AbstractStream], List[Stream]]:
         concurrent_streams: List[AbstractStream] = []
         synchronous_streams: List[Stream] = []
@@ -207,7 +206,7 @@ def _group_streams(
 
         name_to_stream_mapping = {stream["name"]: stream for stream in streams}
 
-        for declarative_stream in self.streams(config=config, catalog=catalog):
+        for declarative_stream in self.streams(config=config):
             # 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

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -20,7 +20,6 @@
     AirbyteMessage,
     AirbyteStateMessage,
     ConfiguredAirbyteCatalog,
-    ConfiguredAirbyteStream,
     ConnectorSpecification,
     FailureType,
 )
@@ -93,6 +92,7 @@ def __init__(
         emit_connector_builder_messages: bool = False,
         component_factory: Optional[ModelToComponentFactory] = None,
         normalize_manifest: Optional[bool] = False,
+        catalog: Optional[ConfiguredAirbyteCatalog] = None,
     ) -> None:
         """
         Args:
@@ -119,6 +119,7 @@ def __init__(
             else ModelToComponentFactory(
                 emit_connector_builder_messages,
                 max_concurrent_async_job_count=source_config.get("max_concurrent_async_job_count"),
+                catalog=catalog,
             )
         )
         self._message_repository = self._constructor.get_message_repository()
@@ -230,9 +231,7 @@ def connection_checker(self) -> ConnectionChecker:
                 f"Expected to generate a ConnectionChecker component, but received {check_stream.__class__}"
             )
 
-    def streams(
-        self, config: Mapping[str, Any], catalog: Optional[ConfiguredAirbyteCatalog] = None
-    ) -> List[Stream]:
+    def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         self._emit_manifest_debug_message(
             extra_args={
                 "source_name": self.name,
@@ -248,7 +247,6 @@ def streams(
         if api_budget_model:
             self._constructor.set_api_budget(api_budget_model, config)
 
-        catalog_with_streams_name = self._catalog_with_streams_name(catalog)
         source_streams = [
             self._constructor.create_component(
                 (
@@ -259,45 +257,12 @@ def streams(
                 stream_config,
                 config,
                 emit_connector_builder_messages=self._emit_connector_builder_messages,
-                include_files=self._get_include_files(
-                    stream_config=stream_config, catalog_with_streams_name=catalog_with_streams_name
-                ),
             )
             for stream_config in self._initialize_cache_for_parent_streams(deepcopy(stream_configs))
         ]
 
         return source_streams
 
-    @staticmethod
-    def _get_include_files(
-        stream_config: Dict[str, Any],
-        catalog_with_streams_name: Mapping[str, ConfiguredAirbyteStream] | None,
-    ) -> bool:
-        """
-        Returns the include_files for the stream if it exists in the catalog.
-        """
-        if catalog_with_streams_name:
-            stream_name = stream_config.get("name")
-            configured_catalog_stream = (
-                catalog_with_streams_name.get(stream_name) if stream_name else None
-            )
-            return bool(configured_catalog_stream and configured_catalog_stream.include_files)
-        return False
-
-    @staticmethod
-    def _catalog_with_streams_name(
-        catalog: ConfiguredAirbyteCatalog | None,
-    ) -> Mapping[str, ConfiguredAirbyteStream] | None:
-        """
-        Returns a dict mapping stream names to their corresponding ConfiguredAirbyteStream objects.
-        """
-        if catalog:
-            return {
-                configured_stream.stream.name: configured_stream
-                for configured_stream in catalog.streams
-            }
-        return None
-
     @staticmethod
     def _initialize_cache_for_parent_streams(
         stream_configs: List[Dict[str, Any]],

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -27,7 +27,7 @@
 from isodate import parse_duration
 from pydantic.v1 import BaseModel
 
-from airbyte_cdk.models import FailureType, Level
+from airbyte_cdk.models import ConfiguredAirbyteCatalog, ConfiguredAirbyteStream, FailureType, Level
 from airbyte_cdk.sources.connector_state_manager import ConnectorStateManager
 from airbyte_cdk.sources.declarative.async_job.job_orchestrator import AsyncJobOrchestrator
 from airbyte_cdk.sources.declarative.async_job.job_tracker import JobTracker
@@ -576,6 +576,7 @@ def __init__(
         disable_retries: bool = False,
         disable_cache: bool = False,
         disable_resumable_full_refresh: bool = False,
+        catalog: ConfiguredAirbyteCatalog = ConfiguredAirbyteCatalog(streams=[]),
         message_repository: Optional[MessageRepository] = None,
         connector_state_manager: Optional[ConnectorStateManager] = None,
         max_concurrent_async_job_count: Optional[int] = None,
@@ -593,6 +594,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)
+        self._catalog_with_streams_name = self._get_catalog_with_streams_name(catalog)
 
     def _init_mappings(self) -> None:
         self.PYDANTIC_MODEL_TO_CONSTRUCTOR: Mapping[Type[BaseModel], Callable[..., Any]] = {
@@ -1852,7 +1854,7 @@ def create_declarative_stream(
                 )
         file_uploader = None
         if model.file_uploader:
-            include_files = kwargs.pop("include_files", False)
+            include_files = self._get_include_files(model)
             file_uploader = self._create_component_from_model(
                 model=model.file_uploader, config=config, include_files=include_files
             )
@@ -3711,3 +3713,27 @@ def create_grouping_partition_router(
             deduplicate=model.deduplicate if model.deduplicate is not None else True,
             config=config,
         )
+
+    @staticmethod
+    def _get_catalog_with_streams_name(
+        catalog: ConfiguredAirbyteCatalog,
+    ) -> Mapping[str, ConfiguredAirbyteStream]:
+        """
+        Returns a dict mapping stream names to their corresponding ConfiguredAirbyteStream objects.
+        """
+        return {
+            configured_stream.stream.name: configured_stream
+            for configured_stream in catalog.streams
+        }
+
+    def _get_include_files(
+        self,
+        stream_model: DeclarativeStreamModel,
+    ) -> bool:
+        """
+        Returns the include_files for the stream if it exists in the catalog.
+        """
+        if stream_model.name and self._catalog_with_streams_name:
+            configured_catalog_stream = self._catalog_with_streams_name.get(stream_model.name)
+            return bool(configured_catalog_stream and configured_catalog_stream.include_files)
+        return False