updated

Author: Oleksandr Bazarnov

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -125,7 +125,7 @@ def dynamic_streams(self) -> List[Dict[str, Any]]:
         )
 
     def deprecation_warnings(self) -> List[AirbyteLogMessage]:
-        return self._constructor.get_model_deprecations() or []
+        return self._constructor.get_model_deprecations()
 
     @property
     def connection_checker(self) -> ConnectionChecker:

airbyte_cdk/sources/declarative/models/base_model_with_deprecations.py

@@ -31,8 +31,8 @@ class BaseModelWithDeprecations(BaseModel):
     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,
+    The `_deprecation_logs` attribute is stored in the model itself.
+    The collected deprecation warnings are further propagated 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,
@@ -46,14 +46,14 @@ class Config:
 
         extra = "allow"
 
-    _deprecation_logs: List[AirbyteLogMessage] = []
-
     def __init__(self, **data: Any) -> None:
         """
         Show warnings for deprecated fields during component initialization.
         """
-        model_fields = self.__fields__
 
+        self._deprecation_logs: List[AirbyteLogMessage] = []
+
+        model_fields = self.__fields__
         for field_name in data:
             if field_name in model_fields:
                 is_deprecated_field = model_fields[field_name].field_info.extra.get(
@@ -74,7 +74,6 @@ def __getattribute__(self, name: str) -> Any:
         """
 
         value = super().__getattribute__(name)
-
         try:
             model_fields = super().__getattribute__(FIELDS_TAG)
             field_info = model_fields.get(name)

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -589,7 +589,7 @@ def __init__(
         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] = []
+        self._collected_deprecation_logs: List[AirbyteLogMessage] = []
 
     def _init_mappings(self) -> None:
         self.PYDANTIC_MODEL_TO_CONSTRUCTOR: Mapping[Type[BaseModel], Callable[..., Any]] = {
@@ -747,14 +747,22 @@ def get_model_deprecations(self) -> List[Any]:
         """
         Returns the deprecation warnings that were collected during the creation of components.
         """
-        return self._deprecation_logs
+        return self._collected_deprecation_logs
 
     def _collect_model_deprecations(self, model: BaseModelWithDeprecations) -> None:
+        """
+        Collects deprecation logs from the given model and appends any new logs to the internal collection.
+
+        This method checks if the provided model has deprecation logs (identified by the presence of the DEPRECATION_LOGS_TAG attribute and a non-None `_deprecation_logs` property). It iterates through each deprecation log in the model and appends it to the `_collected_deprecation_logs` list if it has not already been collected, ensuring that duplicate logs are avoided.
+
+        Args:
+            model (BaseModelWithDeprecations): The model instance from which to collect deprecation logs.
+        """
         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)
+                if log not in self._collected_deprecation_logs:
+                    self._collected_deprecation_logs.append(log)
 
     @staticmethod
     def create_added_field_definition(