updated the migrations approach

Author: Oleksandr Bazarnov

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1858,20 +1858,6 @@ definitions:
       type:
         enum: [Bearer]
   HttpRequester:
-    migrations:
-      - type: replace_field
-        description: The `url_base` has been deprecated, in favor of the `url` field.
-        original_key: url_base
-        replacement_key: url
-      - type: stack
-        migrations:
-          - type: handle_url_parts
-            description: The `path` has been deprecated, in favor of the `url` field. The value from the `path` field will be joined to the `url` field.
-            original_key: path
-            replacement_key: url
-          - type: remove_field
-            description: The `path` has been deprecated, in favor of the `url` field.
-            original_key: path
     title: HTTP Requester
     description: Requester submitting HTTP requests and extracting records from the response.
     type: object

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -26,6 +26,9 @@
 from airbyte_cdk.sources.declarative.checks import COMPONENTS_CHECKER_TYPE_MAPPING
 from airbyte_cdk.sources.declarative.checks.connection_checker import ConnectionChecker
 from airbyte_cdk.sources.declarative.declarative_source import DeclarativeSource
+from airbyte_cdk.sources.declarative.migrations.manifest.migration_handler import (
+    ManifestMigrationHandler,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     DeclarativeStream as DeclarativeStreamModel,
 )
@@ -39,9 +42,6 @@
 from airbyte_cdk.sources.declarative.parsers.manifest_component_transformer import (
     ManifestComponentTransformer,
 )
-from airbyte_cdk.sources.declarative.parsers.manifest_migration_handler import (
-    ManifestMigrationHandler,
-)
 from airbyte_cdk.sources.declarative.parsers.manifest_reference_resolver import (
     ManifestReferenceResolver,
 )
@@ -114,10 +114,9 @@ def __init__(
             "", resolved_source_config, {}
         )
 
-        # migrate definitions to the new format, if any are present
         migrated_source_config = ManifestMigrationHandler(
-            propagated_source_config, self._declarative_component_schema
-        ).migrate()
+            propagated_source_config
+        ).apply_migrations()
 
         self._source_config = migrated_source_config
         self._debug = debug

airbyte_cdk/sources/declarative/migrations/manifest/__init__.py

@@ -0,0 +1,75 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+
+from abc import abstractmethod
+from typing import Any, Dict
+
+ManifestType = Dict[str, Any]
+
+
+TYPE_TAG = "type"
+
+NON_MIGRATABLE_TYPES = [
+    "DynamicDeclarativeStream",
+]
+
+
+class ManifestMigration:
+    @abstractmethod
+    def should_migrate(self, manifest: ManifestType) -> bool:
+        """
+        Check if the manifest should be migrated.
+
+        :param manifest: The manifest to potentially migrate
+        :param kwargs: Additional arguments for migration
+
+        :return: true if the manifest is of the expected format and should be migrated. False otherwise.
+        """
+
+    @abstractmethod
+    def migrate(self, manifest: ManifestType) -> None:
+        """
+        Migrate the manifest. Assumes should_migrate(manifest) returned True.
+
+        :param manifest: The manifest to migrate
+        :param kwargs: Additional arguments for migration
+        """
+
+    def _process_manifest(self, obj: Any) -> None:
+        """
+        Recursively processes a manifest object, migrating components that match the migration criteria.
+
+        This method traverses the entire manifest structure (dictionaries and lists) and applies
+        migrations to components that:
+        1. Have a type tag
+        2. Are not in the list of non-migratable types
+        3. Meet the conditions defined in the should_migrate method
+
+        Parameters:
+            obj (Any): The object to process, which can be a dictionary, list, or any other type.
+                       Dictionary objects are checked for component type tags and potentially migrated.
+                       List objects have each of their items processed recursively.
+                       Other types are ignored.
+
+        Returns:
+            None, since we process the manifest in place.
+        """
+        if isinstance(obj, dict):
+            obj_keys = obj.keys()
+            # check for component type match the designed migration
+            if TYPE_TAG in obj_keys:
+                obj_type = obj[TYPE_TAG]
+
+                # do not migrate if the particular type is in the list of non-migratable types
+                if obj_type in NON_MIGRATABLE_TYPES:
+                    return
+
+                if self.should_migrate(obj):
+                    self.migrate(obj)
+
+            # Process all values in the dictionary
+            for v in list(obj.values()):
+                self._process_manifest(v)
+        elif isinstance(obj, list):
+            # Process all items in the list
+            for item in obj:
+                self._process_manifest(item)

airbyte_cdk/sources/declarative/migrations/manifest/manifest_migration.py

@@ -0,0 +1,62 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+
+import copy
+from typing import Type
+
+from airbyte_cdk.sources.declarative.migrations.manifest.manifest_migration import (
+    ManifestMigration,
+    ManifestType,
+)
+from airbyte_cdk.sources.declarative.migrations.manifest.migrations_registry import (
+    migrations_registry,
+)
+from airbyte_cdk.sources.declarative.parsers.custom_exceptions import (
+    ManifestMigrationException,
+)
+
+
+class ManifestMigrationHandler:
+    """
+    This class is responsible for handling migrations in the manifest.
+    """
+
+    def __init__(self, manifest: ManifestType) -> None:
+        self._manifest = manifest
+        self._migrated_manifest: ManifestType = copy.deepcopy(self._manifest)
+
+    def apply_migrations(self) -> ManifestType:
+        """
+        Apply all registered migrations to the manifest.
+
+        This method iterates through all migrations in the migrations registry and applies
+        them sequentially to the current manifest. If any migration fails with a
+        ManifestMigrationException, the original unmodified manifest is returned instead.
+
+        Returns:
+            ManifestType: The migrated manifest if all migrations succeeded, or the original
+                          manifest if any migration failed.
+        """
+        try:
+            for migration_class in migrations_registry:
+                self._handle_migration(migration_class)
+            return self._migrated_manifest
+        except ManifestMigrationException:
+            # if any errors occur we return the original resolved manifest
+            return self._manifest
+
+    def _handle_migration(self, migration_class: Type[ManifestMigration]) -> None:
+        """
+        Handles a single manifest migration by instantiating the migration class and processing the manifest.
+
+        Args:
+            migration_class (Type[ManifestMigration]): The migration class to apply to the manifest.
+
+        Raises:
+            ManifestMigrationException: If the migration process encounters any errors.
+        """
+        try:
+            migration_class()._process_manifest(self._migrated_manifest)
+        except Exception as e:
+            raise ManifestMigrationException(f"Failed to migrate the manifest: {e}") from e

airbyte_cdk/sources/declarative/migrations/manifest/migration_handler.py

@@ -0,0 +1,11 @@
+from airbyte_cdk.sources.declarative.migrations.manifest.migrations.http_requester_path_to_url_migration import (
+    HttpRequesterPathToUrlMigration,
+)
+from airbyte_cdk.sources.declarative.migrations.manifest.migrations.http_requester_url_base_to_url_migration import (
+    HttpRequesterUrlBaseToUrlMigration,
+)
+
+__all__ = [
+    "HttpRequesterUrlBaseToUrlMigration",
+    "HttpRequesterPathToUrlMigration",
+]

airbyte_cdk/sources/declarative/migrations/manifest/migrations/__init__.py

@@ -0,0 +1,36 @@
+from urllib.parse import urljoin
+
+from airbyte_cdk.sources.declarative.migrations.manifest.manifest_migration import (
+    TYPE_TAG,
+    ManifestMigration,
+    ManifestType,
+)
+from airbyte_cdk.sources.types import EmptyString
+
+
+class HttpRequesterPathToUrlMigration(ManifestMigration):
+    component_type = "HttpRequester"
+    original_key = "path"
+    replacement_key = "url"
+
+    def should_migrate(self, manifest: ManifestType) -> bool:
+        return manifest[TYPE_TAG] == self.component_type and self.original_key in list(
+            manifest.keys()
+        )
+
+    def migrate(self, manifest: ManifestType) -> None:
+        original_key_value = manifest[self.original_key].lstrip("/")
+        replacement_key_value = manifest[self.replacement_key]
+
+        # return a full-url if provided directly from interpolation context
+        if original_key_value == EmptyString or original_key_value is None:
+            manifest[self.replacement_key] = replacement_key_value
+            manifest.pop(self.original_key, None)
+        else:
+            # since we didn't provide a full-url, the url_base might not have a trailing slash
+            # so we join the url_base and path correctly
+            if not replacement_key_value.endswith("/"):
+                replacement_key_value += "/"
+
+            manifest[self.replacement_key] = urljoin(replacement_key_value, original_key_value)
+            manifest.pop(self.original_key, None)

airbyte_cdk/sources/declarative/migrations/manifest/migrations/http_requester_path_to_url_migration.py

@@ -0,0 +1,20 @@
+from airbyte_cdk.sources.declarative.migrations.manifest.manifest_migration import (
+    TYPE_TAG,
+    ManifestMigration,
+    ManifestType,
+)
+
+
+class HttpRequesterUrlBaseToUrlMigration(ManifestMigration):
+    component_type = "HttpRequester"
+    original_key = "url_base"
+    replacement_key = "url"
+
+    def should_migrate(self, manifest: ManifestType) -> bool:
+        return manifest[TYPE_TAG] == self.component_type and self.original_key in list(
+            manifest.keys()
+        )
+
+    def migrate(self, manifest: ManifestType) -> None:
+        manifest[self.replacement_key] = manifest[self.original_key]
+        manifest.pop(self.original_key, None)

airbyte_cdk/sources/declarative/migrations/manifest/migrations/http_requester_url_base_to_url_migration.py

@@ -0,0 +1,17 @@
+from typing import List, Type
+
+from airbyte_cdk.sources.declarative.migrations.manifest.manifest_migration import (
+    ManifestMigration,
+)
+from airbyte_cdk.sources.declarative.migrations.manifest.migrations import (
+    HttpRequesterPathToUrlMigration,
+    HttpRequesterUrlBaseToUrlMigration,
+)
+
+# This is the registry of all the migrations that are available.
+# Add new migrations to the bottom of the list,
+# ( ! ) make sure the order of the migrations is correct.
+migrations_registry: List[Type[ManifestMigration]] = [
+    HttpRequesterUrlBaseToUrlMigration,
+    HttpRequesterPathToUrlMigration,
+]

airbyte_cdk/sources/declarative/migrations/manifest/migrations_registry.py

@@ -8,6 +8,13 @@
 
 from pydantic.v1 import BaseModel
 
+from airbyte_cdk.models import (
+    AirbyteLogMessage,
+    AirbyteMessage,
+    Level,
+    Type,
+)
+
 # format the warning message
 warnings.formatwarning = (
     lambda message, category, *args, **kwargs: f"{category.__name__}: {message}"
@@ -36,6 +43,16 @@ def _deprecated_warning(self, field_name: str, message: str) -> None:
             DeprecationWarning,
         )
 
+        # print(
+        #     AirbyteMessage(
+        #         type=Type.LOG,
+        #         log=AirbyteLogMessage(
+        #             level=Level.WARN,
+        #             message=f"Component type: `{self.__class__.__name__}`. Field '{field_name}' is deprecated. {message}",
+        #         ),
+        #     )
+        # )
+
     def __init__(self, **data: Any) -> None:
         """
         Show warnings for deprecated fields during component initialization.