emit deprecation warnings in Connector Builder

Author: Oleksandr Bazarnov

airbyte_cdk/connector_builder/test_reader/reader.py

@@ -110,11 +110,16 @@ def run_test_read(
         record_limit = self._check_record_limit(record_limit)
         # The connector builder currently only supports reading from a single stream at a time
         stream = source.streams(config)[0]
+
+        # get any deprecation warnings during the component creation
+        deprecation_warnings: List[AirbyteLogMessage] = source.deprecation_warnings()
+
         schema_inferrer = SchemaInferrer(
             self._pk_to_nested_and_composite_field(stream.primary_key),
             self._cursor_field_to_nested_and_composite_field(stream.cursor_field),
         )
         datetime_format_inferrer = DatetimeFormatInferrer()
+
         message_group = get_message_groups(
             self._read_stream(source, config, configured_catalog, state),
             schema_inferrer,
@@ -123,7 +128,7 @@ def run_test_read(
         )
 
         slices, log_messages, auxiliary_requests, latest_config_update = self._categorise_groups(
-            message_group
+            message_group, deprecation_warnings
         )
         schema, log_messages = self._get_infered_schema(
             configured_catalog, schema_inferrer, log_messages
@@ -236,7 +241,11 @@ def _check_record_limit(self, record_limit: Optional[int] = None) -> int:
 
         return record_limit
 
-    def _categorise_groups(self, message_groups: MESSAGE_GROUPS) -> GROUPED_MESSAGES:
+    def _categorise_groups(
+        self,
+        message_groups: MESSAGE_GROUPS,
+        deprecation_warnings: Optional[List[Any]] = None,
+    ) -> GROUPED_MESSAGES:
         """
         Categorizes a sequence of message groups into slices, log messages, auxiliary requests, and the latest configuration update.
 
@@ -267,6 +276,7 @@ def _categorise_groups(self, message_groups: MESSAGE_GROUPS) -> GROUPED_MESSAGES
         auxiliary_requests = []
         latest_config_update: Optional[AirbyteControlMessage] = None
 
+        # process the message groups first
         for message_group in message_groups:
             match message_group:
                 case AirbyteLogMessage():
@@ -296,6 +306,17 @@ def _categorise_groups(self, message_groups: MESSAGE_GROUPS) -> GROUPED_MESSAGES
                 case _:
                     raise ValueError(f"Unknown message group type: {type(message_group)}")
 
+        # process deprecation warnings, if present
+        if deprecation_warnings is not None:
+            for deprecation in deprecation_warnings:
+                match deprecation:
+                    case AirbyteLogMessage():
+                        log_messages.append(
+                            LogMessage(message=deprecation.message, level=deprecation.level.value)
+                        )
+                    case _:
+                        raise ValueError(f"Unknown message group type: {type(deprecation)}")
+
         return slices, log_messages, auxiliary_requests, latest_config_update
 
     def _get_infered_schema(

airbyte_cdk/sources/declarative/declarative_source.py

@@ -4,8 +4,11 @@
 
 import logging
 from abc import abstractmethod
-from typing import Any, Mapping, Tuple
+from typing import Any, List, Mapping, Tuple
 
+from airbyte_cdk.models import (
+    AirbyteLogMessage,
+)
 from airbyte_cdk.sources.abstract_source import AbstractSource
 from airbyte_cdk.sources.declarative.checks.connection_checker import ConnectionChecker
 
@@ -34,3 +37,9 @@ def check_connection(
           The error object will be cast to string to display the problem to the user.
         """
         return self.connection_checker.check_connection(self, logger, config)
+
+    def deprecation_warnings(self) -> List[AirbyteLogMessage]:
+        """
+        Returns a list of deprecation warnings for the source.
+        """
+        return []

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -17,6 +17,7 @@
 
 from airbyte_cdk.models import (
     AirbyteConnectionStatus,
+    AirbyteLogMessage,
     AirbyteMessage,
     AirbyteStateMessage,
     ConfiguredAirbyteCatalog,
@@ -60,24 +61,6 @@
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 
 
-def _get_declarative_component_schema() -> Dict[str, Any]:
-    try:
-        raw_component_schema = pkgutil.get_data(
-            "airbyte_cdk", "sources/declarative/declarative_component_schema.yaml"
-        )
-        if raw_component_schema is not None:
-            declarative_component_schema = yaml.load(raw_component_schema, Loader=yaml.SafeLoader)
-            return declarative_component_schema  # type: ignore
-        else:
-            raise RuntimeError(
-                "Failed to read manifest component json schema required for deduplication"
-            )
-    except FileNotFoundError as e:
-        raise FileNotFoundError(
-            f"Failed to read manifest component json schema required for deduplication: {e}"
-        )
-
-
 class ManifestDeclarativeSource(DeclarativeSource):
     """Declarative source defined by a manifest of low-code components that define source connector behavior"""
 
@@ -100,7 +83,6 @@ def __init__(
         """
         self.logger = logging.getLogger(f"airbyte.{self.name}")
 
-        self._declarative_component_schema = _get_declarative_component_schema()
         # For ease of use we don't require the type to be specified at the top level manifest, but it should be included during processing
         manifest = dict(source_config)
         if "type" not in manifest:
@@ -151,6 +133,9 @@ def dynamic_streams(self) -> List[Dict[str, Any]]:
             manifest=self._source_config, config=self._config, with_dynamic_stream_name=True
         )
 
+    def deprecation_warnings(self) -> List[AirbyteLogMessage]:
+        return self._constructor.get_model_deprecations() or []
+
     @property
     def connection_checker(self) -> ConnectionChecker:
         check = self._source_config["check"]

airbyte_cdk/sources/declarative/models/base_model_with_deprecations.py

@@ -4,15 +4,13 @@
 # WHEN DEPRECATED FIELDS ARE ACCESSED
 
 import warnings
-from typing import Any
+from typing import Any, List
 
 from pydantic.v1 import BaseModel
 
 from airbyte_cdk.models import (
     AirbyteLogMessage,
-    AirbyteMessage,
     Level,
-    Type,
 )
 
 # format the warning message
@@ -23,48 +21,49 @@
 FIELDS_TAG = "__fields__"
 DEPRECATED = "deprecated"
 DEPRECATION_MESSAGE = "deprecation_message"
+DEPRECATION_LOGS_TAG = "_deprecation_logs"
 
 
 class BaseModelWithDeprecations(BaseModel):
     """
     Pydantic BaseModel that warns when deprecated fields are accessed.
+    The deprecation message is stored in the field's extra attributes.
+    This class is used to create models that can have deprecated fields
+    and show warnings when those fields are accessed or initialized.
+
+    The `_deprecation_logs` attribute is storred in the model itself.
+    The collected deprecation warnings are further proparated to the Airbyte log messages,
+    during the component creation process, in `model_to_component._collect_model_deprecations()`.
+
+    The component implementation is not responsible for handling the deprecation warnings,
+    since the deprecation warnings are already handled in the model itself.
     """
 
-    def _deprecated_warning(self, field_name: str, message: str) -> None:
+    class Config:
         """
-        Show a warning message for deprecated fields (to stdout).
-        Args:
-            field_name (str): Name of the deprecated field.
-            message (str): Warning message to be displayed.
+        Allow extra fields in the model. In case the model restricts extra fields.
         """
 
-        warnings.warn(
-            f"Component type: `{self.__class__.__name__}`. Field '{field_name}' is deprecated. {message}",
-            DeprecationWarning,
-        )
+        extra = "allow"
 
-        # print(
-        #     AirbyteMessage(
-        #         type=Type.LOG,
-        #         log=AirbyteLogMessage(
-        #             level=Level.WARN,
-        #             message=f"Component type: `{self.__class__.__name__}`. Field '{field_name}' is deprecated. {message}",
-        #         ),
-        #     )
-        # )
+    _deprecation_logs: List[AirbyteLogMessage] = []
 
     def __init__(self, **data: Any) -> None:
         """
         Show warnings for deprecated fields during component initialization.
         """
-
         model_fields = self.__fields__
 
         for field_name in data:
             if field_name in model_fields:
-                if model_fields[field_name].field_info.extra.get(DEPRECATED, False):
-                    message = model_fields[field_name].field_info.extra.get(DEPRECATION_MESSAGE, "")
-                    self._deprecated_warning(field_name, message)
+                is_deprecated_field = model_fields[field_name].field_info.extra.get(
+                    DEPRECATED, False
+                )
+                if is_deprecated_field:
+                    deprecation_message = model_fields[field_name].field_info.extra.get(
+                        DEPRECATION_MESSAGE, ""
+                    )
+                    self._deprecated_warning(field_name, deprecation_message)
 
         # Call the parent constructor
         super().__init__(**data)
@@ -76,13 +75,39 @@ def __getattribute__(self, name: str) -> Any:
 
         value = super().__getattribute__(name)
 
-        if name == FIELDS_TAG:
-            try:
-                model_fields = super().__getattribute__(FIELDS_TAG)
-                field_info = model_fields.get(name)
-                if field_info and field_info.field_info.extra.get(DEPRECATED):
-                    self._deprecated_warning(name, field_info)
-            except (AttributeError, KeyError):
-                pass
+        try:
+            model_fields = super().__getattribute__(FIELDS_TAG)
+            field_info = model_fields.get(name)
+            is_deprecated_field = (
+                field_info.field_info.extra.get(DEPRECATED, False) if field_info else False
+            )
+            if is_deprecated_field:
+                deprecation_message = field_info.extra.get(DEPRECATION_MESSAGE, "")
+                self._deprecated_warning(name, deprecation_message)
+        except (AttributeError, KeyError):
+            pass
 
         return value
+
+    def _deprecated_warning(self, field_name: str, message: str) -> None:
+        """
+        Show a warning message for deprecated fields (to stdout).
+        Args:
+            field_name (str): Name of the deprecated field.
+            message (str): Warning message to be displayed.
+        """
+
+        # Emit a warning message for deprecated fields (to stdout) (Python Default behavior)
+        warnings.warn(
+            f"Component type: `{self.__class__.__name__}`. Field '{field_name}' is deprecated. {message}",
+            DeprecationWarning,
+        )
+
+        # Add the deprecation message to the Airbyte log messages,
+        # this logs are displayed in the Connector Builder.
+        self._deprecation_logs.append(
+            AirbyteLogMessage(
+                level=Level.WARN,
+                message=f"Component type: `{self.__class__.__name__}`. Field '{field_name}' is deprecated. {message}",
+            ),
+        )

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 AirbyteLogMessage, 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
@@ -108,6 +108,10 @@
     CustomStateMigration,
     GzipDecoder,
 )
+from airbyte_cdk.sources.declarative.models.base_model_with_deprecations import (
+    DEPRECATION_LOGS_TAG,
+    BaseModelWithDeprecations,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     AddedFieldDefinition as AddedFieldDefinitionModel,
 )
@@ -583,6 +587,8 @@ 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)
+        # placeholder for deprecation warnings
+        self._deprecation_logs: List[AirbyteLogMessage] = []
 
     def _init_mappings(self) -> None:
         self.PYDANTIC_MODEL_TO_CONSTRUCTOR: Mapping[Type[BaseModel], Callable[..., Any]] = {
@@ -729,8 +735,26 @@ def _create_component_from_model(self, model: BaseModel, config: Config, **kwarg
         component_constructor = self.PYDANTIC_MODEL_TO_CONSTRUCTOR.get(model.__class__)
         if not component_constructor:
             raise ValueError(f"Could not find constructor for {model.__class__}")
+
+        # collect deprecation warnings for supported models.
+        if isinstance(model, BaseModelWithDeprecations):
+            self._collect_model_deprecations(model)
+
         return component_constructor(model=model, config=config, **kwargs)
 
+    def get_model_deprecations(self) -> List[Any]:
+        """
+        Returns the deprecation warnings that were collected during the creation of components.
+        """
+        return self._deprecation_logs
+
+    def _collect_model_deprecations(self, model: BaseModelWithDeprecations) -> None:
+        if hasattr(model, DEPRECATION_LOGS_TAG) and model._deprecation_logs is not None:
+            for log in model._deprecation_logs:
+                # avoid duplicates for deprecation logs observed.
+                if log not in self._deprecation_logs:
+                    self._deprecation_logs.append(log)
+
     @staticmethod
     def create_added_field_definition(
         model: AddedFieldDefinitionModel, config: Config, **kwargs: Any

unit_tests/connector_builder/test_connector_builder_handler.py

@@ -7,7 +7,7 @@
 import json
 import logging
 import os
-from typing import Literal
+from typing import List, Literal
 from unittest import mock
 from unittest.mock import MagicMock, patch
 
@@ -817,6 +817,9 @@ def spec(self, logger: logging.Logger) -> ConnectorSpecification:
             connector_specification.connectionSpecification = {}
             return connector_specification
 
+        def deprecation_warnings(self) -> List[AirbyteLogMessage]:
+            return []
+
         @property
         def check_config_against_spec(self) -> Literal[False]:
             return False