add deprecation warnings

Author: Oleksandr Bazarnov

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1883,6 +1883,7 @@ definitions:
         enum: [HttpRequester]
       url_base:
         deprecated: true
+        deprecation_message: "Use `url` field instead."
         sharable: true
         title: API Base URL
         description: Deprecated, use the `url` instead. Base URL of the API source. Do not put sensitive information (e.g. API tokens) into this field - Use the Authentication component for this.
@@ -1922,6 +1923,7 @@ definitions:
           - "https://example.com/api/v1/resource/{{ next_page_token['id'] }}"
       path:
         deprecated: true
+        deprecation_message: "Use `url` field instead."
         title: URL Path
         description: Deprecated, use the `url` instead. Path the specific API endpoint that this stream represents. Do not put sensitive information (e.g. API tokens) into this field - Use the Authentication component for this.
         type: string

airbyte_cdk/sources/declarative/models/base_model_with_deprecations.py

@@ -0,0 +1,71 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+
+# THIS IS A STATIC CLASS MODEL USED TO DISPLAY DEPRECATION WARNINGS
+# WHEN DEPRECATED FIELDS ARE ACCESSED
+
+import warnings
+from typing import Any
+
+from pydantic.v1 import BaseModel
+
+# format the warning message
+warnings.formatwarning = (
+    lambda message, category, *args, **kwargs: f"{category.__name__}: {message}"
+)
+
+FIELDS_TAG = "__fields__"
+DEPRECATED = "deprecated"
+DEPRECATION_MESSAGE = "deprecation_message"
+
+
+class BaseModelWithDeprecations(BaseModel):
+    """
+    Pydantic BaseModel that warns when deprecated fields are accessed.
+    """
+
+    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.
+        """
+
+        warnings.warn(
+            f"Component type: `{self.__class__.__name__}`. Field '{field_name}' is deprecated. {message}",
+            DeprecationWarning,
+        )
+
+    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)
+
+        # Call the parent constructor
+        super().__init__(**data)
+
+    def __getattribute__(self, name: str) -> Any:
+        """
+        Show warnings for deprecated fields during field usage.
+        """
+
+        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
+
+        return value

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -8,6 +8,10 @@
 
 from pydantic.v1 import BaseModel, Extra, Field
 
+from airbyte_cdk.sources.declarative.models.base_model_with_deprecations import (
+    BaseModelWithDeprecations,
+)
+
 
 class AuthFlowType(Enum):
     oauth2_0 = "oauth2.0"
@@ -2145,11 +2149,12 @@ class SessionTokenAuthenticator(BaseModel):
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
-class HttpRequester(BaseModel):
+class HttpRequester(BaseModelWithDeprecations):
     type: Literal["HttpRequester"]
     url_base: Optional[str] = Field(
         None,
         deprecated=True,
+        deprecation_message="Use `url` field instead.",
         description="Deprecated, use the `url` instead. Base URL of the API source. Do not put sensitive information (e.g. API tokens) into this field - Use the Authentication component for this.",
         examples=[
             "https://connect.squareup.com/v2",
@@ -2173,6 +2178,7 @@ class HttpRequester(BaseModel):
     path: Optional[str] = Field(
         None,
         deprecated=True,
+        deprecation_message="Use `url` field instead.",
         description="Deprecated, use the `url` instead. Path the specific API endpoint that this stream represents. Do not put sensitive information (e.g. API tokens) into this field - Use the Authentication component for this.",
         examples=[
             "/products",

bin/generate_component_manifest_files.py

@@ -1,5 +1,6 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
 
+import re
 import sys
 from glob import glob
 from pathlib import Path
@@ -28,6 +29,63 @@ def generate_init_module_content() -> str:
     return header
 
 
+def replace_base_model_for_classes_with_deprecated_fields(post_processed_content: str) -> str:
+    """
+    Replace the base model for classes with deprecated fields.
+    This function looks for classes that inherit from `BaseModel` and have fields marked as deprecated.
+    It replaces the base model with `BaseModelWithDeprecations` for those classes.
+    """
+
+    # Find classes with deprecated fields
+    classes_with_deprecated_fields = set()
+    class_matches = re.finditer(r"class (\w+)\(BaseModel\):", post_processed_content)
+
+    for class_match in class_matches:
+        class_name = class_match.group(1)
+        class_start = class_match.start()
+        # Find the next class definition or end of file
+        next_class_match = re.search(
+            r"class \w+\(",
+            post_processed_content[class_start + len(class_match.group(0)) :],
+        )
+        class_end = (
+            len(post_processed_content)
+            if next_class_match is None
+            else class_start + len(class_match.group(0)) + next_class_match.start()
+        )
+        class_content = post_processed_content[class_start:class_end]
+
+        # Check if any field has deprecated=True
+        if re.search(r"deprecated\s*=\s*True", class_content):
+            classes_with_deprecated_fields.add(class_name)
+
+    # update the imports to include the new base model with deprecation warinings
+    # only if there are classes with the fields marked as deprecated.
+    if len(classes_with_deprecated_fields) > 0:
+        # Find where to insert the base model - after imports but before class definitions
+        imports_end = post_processed_content.find(
+            "\n\n",
+            post_processed_content.find("from pydantic.v1 import"),
+        )
+        if imports_end > 0:
+            post_processed_content = (
+                post_processed_content[:imports_end]
+                + "\n\n"
+                + "from airbyte_cdk.sources.declarative.models.base_model_with_deprecations import (\n"
+                + "    BaseModelWithDeprecations,\n"
+                + ")"
+                + post_processed_content[imports_end:]
+            )
+
+    # Use the `BaseModelWithDeprecations` base model for the classes with deprecated fields
+    for class_name in classes_with_deprecated_fields:
+        pattern = rf"class {class_name}\(BaseModel\):"
+        replacement = f"class {class_name}(BaseModelWithDeprecations):"
+        post_processed_content = re.sub(pattern, replacement, post_processed_content)
+
+    return post_processed_content
+
+
 async def post_process_codegen(codegen_container: dagger.Container):
     codegen_container = codegen_container.with_exec(
         ["mkdir", "/generated_post_processed"], use_entrypoint=True
@@ -41,6 +99,11 @@ async def post_process_codegen(codegen_container: dagger.Container):
             post_processed_content = original_content.replace(
                 " _parameters:", " parameters:"
             ).replace("from pydantic", "from pydantic.v1")
+
+            post_processed_content = replace_base_model_for_classes_with_deprecated_fields(
+                post_processed_content
+            )
+
             codegen_container = codegen_container.with_new_file(
                 f"/generated_post_processed/{generated_file}", contents=post_processed_content
             )
@@ -75,9 +138,12 @@ async def main():
                     "--set-default-enum-member",
                     "--use-double-quotes",
                     "--remove-special-field-name-prefix",
-                    # account the `deprecated` flag provided for the field.
+                    # allow usage of the extra key such as `deprecated`, etc.
                     "--field-extra-keys",
+                    # account the `deprecated` flag provided for the field.
                     "deprecated",
+                    # account the `deprecation_message` provided for the field.
+                    "deprecation_message",
                 ],
                 use_entrypoint=True,
             )