airbyte_cdk.sources.declarative.models.base_model_with_deprecations

  1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  2
  3# THIS IS A STATIC CLASS MODEL USED TO DISPLAY DEPRECATION WARNINGS
  4# WHEN DEPRECATED FIELDS ARE ACCESSED
  5
  6import warnings
  7
  8# ignore the SyntaxWarning in the Airbyte log messages, during the string evaluation
  9warnings.filterwarnings("ignore", category=SyntaxWarning)
 10
 11from typing import Any, List
 12
 13from pydantic.v1 import BaseModel
 14
 15from airbyte_cdk.connector_builder.models import LogMessage as ConnectorBuilderLogMessage
 16
 17# format the warning message
 18warnings.formatwarning = (
 19    lambda message, category, *args, **kwargs: f"{category.__name__}: {message}\n"
 20)
 21
 22
 23FIELDS_TAG = "__fields__"
 24DEPRECATED = "deprecated"
 25DEPRECATION_MESSAGE = "deprecation_message"
 26DEPRECATION_LOGS_TAG = "_deprecation_logs"
 27
 28
 29class BaseModelWithDeprecations(BaseModel):
 30    """
 31    Pydantic BaseModel that warns when deprecated fields are accessed.
 32    The deprecation message is stored in the field's extra attributes.
 33    This class is used to create models that can have deprecated fields
 34    and show warnings when those fields are accessed or initialized.
 35
 36    The `_deprecation_logs` attribute is stored in the model itself.
 37    The collected deprecation warnings are further propagated to the Airbyte log messages,
 38    during the component creation process, in `model_to_component._collect_model_deprecations()`.
 39
 40    The component implementation is not responsible for handling the deprecation warnings,
 41    since the deprecation warnings are already handled in the model itself.
 42    """
 43
 44    class Config:
 45        """
 46        Allow extra fields in the model. In case the model restricts extra fields.
 47        """
 48
 49        extra = "allow"
 50
 51    def __init__(self, **model_fields: Any) -> None:
 52        """
 53        Show warnings for deprecated fields during component initialization.
 54        """
 55        # call the parent constructor first to initialize Pydantic internals
 56        super().__init__(**model_fields)
 57        # set the placeholder for the default deprecation messages
 58        self._default_deprecation_messages: List[str] = []
 59        # set the placeholder for the deprecation logs
 60        self._deprecation_logs: List[ConnectorBuilderLogMessage] = []
 61        # process deprecated fields, if present
 62        self._process_fields(model_fields)
 63        # emit default deprecation messages
 64        self._emit_default_deprecation_messages()
 65        # set the deprecation logs attribute to the model
 66        self._set_deprecation_logs_attr_to_model()
 67
 68    def _is_deprecated_field(self, field_name: str) -> bool:
 69        return (
 70            self.__fields__[field_name].field_info.extra.get(DEPRECATED, False)
 71            if field_name in self.__fields__.keys()
 72            else False
 73        )
 74
 75    def _get_deprecation_message(self, field_name: str) -> str:
 76        return (
 77            self.__fields__[field_name].field_info.extra.get(
 78                DEPRECATION_MESSAGE, "<missing_deprecation_message>"
 79            )
 80            if field_name in self.__fields__.keys()
 81            else "<missing_deprecation_message>"
 82        )
 83
 84    def _process_fields(self, model_fields: Any) -> None:
 85        """
 86        Processes the fields in the provided model data, checking for deprecated fields.
 87
 88        For each field in the input `model_fields`, this method checks if the field exists in the model's defined fields.
 89        If the field is marked as deprecated (using the `DEPRECATED` flag in its metadata), it triggers a deprecation warning
 90        by calling the `_create_warning` method with the field name and an optional deprecation message.
 91
 92        Args:
 93            model_fields (Any): The data containing fields to be processed.
 94
 95        Returns:
 96            None
 97        """
 98
 99        if hasattr(self, FIELDS_TAG):
100            for field_name in model_fields.keys():
101                if self._is_deprecated_field(field_name):
102                    self._create_warning(
103                        field_name,
104                        self._get_deprecation_message(field_name),
105                    )
106
107    def _set_deprecation_logs_attr_to_model(self) -> None:
108        """
109        Sets the deprecation logs attribute on the model instance.
110
111        This method attaches the current instance's deprecation logs to the model by setting
112        an attribute named by `DEPRECATION_LOGS_TAG` to the value of `self._deprecation_logs`.
113        This is typically used to track or log deprecated features or configurations within the model.
114
115        Returns:
116            None
117        """
118        setattr(self, DEPRECATION_LOGS_TAG, self._deprecation_logs)
119
120    def _create_warning(self, field_name: str, message: str) -> None:
121        """
122        Show a warning message for deprecated fields (to stdout).
123        Args:
124            field_name (str): Name of the deprecated field.
125            message (str): Warning message to be displayed.
126        """
127
128        deprecated_message = f"Component type: `{self.__class__.__name__}`. Field '{field_name}' is deprecated. {message}"
129
130        if deprecated_message not in self._default_deprecation_messages:
131            # Avoid duplicates in the default deprecation messages
132            self._default_deprecation_messages.append(deprecated_message)
133
134        # Create an Airbyte deprecation log message
135        deprecation_log_message = ConnectorBuilderLogMessage(
136            level="WARN", message=deprecated_message
137        )
138        # Add the deprecation message to the Airbyte log messages,
139        # this logs are displayed in the Connector Builder.
140        if deprecation_log_message not in self._deprecation_logs:
141            # Avoid duplicates in the deprecation logs
142            self._deprecation_logs.append(deprecation_log_message)
143
144    def _emit_default_deprecation_messages(self) -> None:
145        """
146        Emit default deprecation messages for deprecated fields to STDOUT.
147        """
148        for message in self._default_deprecation_messages:
149            warnings.warn(message, DeprecationWarning)