updated version checks and error messaging

Author: Oleksandr Bazarnov

airbyte_cdk/manifest_migrations/manifest_migration.py

@@ -4,6 +4,8 @@
 from abc import abstractmethod
 from typing import Any, Dict
 
+from packaging.version import Version
+
 ManifestType = Dict[str, Any]
 
 
@@ -36,13 +38,13 @@ def migrate(self, manifest: ManifestType) -> None:
         """
 
     @property
-    def migration_version(self) -> str:
+    def migration_version(self) -> Version:
         """
         Get the migration version.
 
-        :return: The migration version as a string
+        :return: The migration version as a Version object
         """
-        return self._get_migration_version()
+        return Version(self._get_migration_version())
 
     def _is_component(self, obj: Dict[str, Any]) -> bool:
         """
@@ -53,18 +55,15 @@ def _is_component(self, obj: Dict[str, Any]) -> bool:
         """
         return TYPE_TAG in obj.keys()
 
-    def _is_migratable(self, obj: Dict[str, Any]) -> bool:
+    def _is_migratable_type(self, obj: Dict[str, Any]) -> bool:
         """
         Check if the object is a migratable component,
         based on the Type of the component and the migration version.
 
         :param obj: The object to check
         :return: True if the object is a migratable component, False otherwise
         """
-        return (
-            obj[TYPE_TAG] not in NON_MIGRATABLE_TYPES
-            and self._get_manifest_version(obj) <= self.migration_version
-        )
+        return obj[TYPE_TAG] not in NON_MIGRATABLE_TYPES
 
     def _process_manifest(self, obj: Any) -> None:
         """
@@ -89,7 +88,7 @@ def _process_manifest(self, obj: Any) -> None:
             # Check if the object is a component
             if self._is_component(obj):
                 # Check if the object is allowed to be migrated
-                if not self._is_migratable(obj):
+                if not self._is_migratable_type(obj):
                     return
 
                 # Check if the object should be migrated
@@ -106,15 +105,6 @@ def _process_manifest(self, obj: Any) -> None:
             for item in obj:
                 self._process_manifest(item)
 
-    def _get_manifest_version(self, manifest: ManifestType) -> str:
-        """
-        Get the manifest version from the manifest.
-
-        :param manifest: The manifest to get the version from
-        :return: The manifest version
-        """
-        return str(manifest.get("version", "0.0.0"))
-
     def _get_migration_version(self) -> str:
         """
         Get the migration version from the class name.

airbyte_cdk/manifest_migrations/migration_handler.py

@@ -3,8 +3,11 @@
 #
 
 import copy
+import logging
 from typing import Type
 
+from packaging.version import Version
+
 from airbyte_cdk.manifest_migrations.exceptions import (
     ManifestMigrationException,
 )
@@ -16,6 +19,8 @@
     MIGRATIONS,
 )
 
+LOGGER = logging.getLogger("airbyte.cdk.manifest_migrations")
+
 
 class ManifestMigrationHandler:
     """
@@ -25,6 +30,7 @@ class ManifestMigrationHandler:
     def __init__(self, manifest: ManifestType) -> None:
         self._manifest = manifest
         self._migrated_manifest: ManifestType = copy.deepcopy(self._manifest)
+        self._manifest_version: Version = self._get_manifest_version()
 
     def apply_migrations(self) -> ManifestType:
         """
@@ -57,6 +63,22 @@ def _handle_migration(self, migration_class: Type[ManifestMigration]) -> None:
             ManifestMigrationException: If the migration process encounters any errors.
         """
         try:
-            migration_class()._process_manifest(self._migrated_manifest)
+            migration_instance = migration_class()
+            # check if the migration is supported for the given manifest version
+            if self._manifest_version <= migration_instance.migration_version:
+                migration_instance._process_manifest(self._migrated_manifest)
+            else:
+                LOGGER.info(
+                    f"Manifest migration: `{migration_class.__name__}` is not supported for the given manifest version `{self._manifest_version}`.",
+                )
         except Exception as e:
-            raise ManifestMigrationException(f"Failed to migrate the manifest: {e}") from e
+            raise ManifestMigrationException(str(e)) from e
+
+    def _get_manifest_version(self) -> Version:
+        """
+        Get the manifest version from the manifest.
+
+        :param manifest: The manifest to get the version from
+        :return: The manifest version
+        """
+        return Version(str(self._migrated_manifest.get("version", "0.0.0")))

airbyte_cdk/manifest_migrations/migrations_registry.py

@@ -17,16 +17,45 @@
 
 
 def _migration_order_key(cls: object) -> int:
+    """
+    Determines the migration order key for a given migration class based on its module name.
+
+    The function expects the module name to end with a double underscore followed by an integer (e.g., '__0').
+    This integer is extracted and returned as the migration order key.
+
+    Args:
+        cls (object): The migration class whose module name encodes the migration order.
+
+    Returns:
+        int: The migration order extracted from the module name.
+
+    Raises:
+        ValueError: If the module name does not contain the expected order suffix.
+    """
     # Extract the migration order from the module name, e.g., http_requester_url_base_to_url_v6_45_2__0
     # The order is the integer after the double underscore at the end of the module name
     module_name = cls.__module__.split(".")[-1]
     match = re.search(r"__(\d+)$", module_name)
-    return int(match.group(1)) if match else 0
+    if match:
+        return int(match.group(1))
+    else:
+        message = f"Migration `{cls.__module__}` doesn't have the `order` in the module name: {module_name}. Did you miss to add `__<order>` to the module name?"
+        raise ValueError(message)
 
 
 def _discover_migrations() -> List[Type[ManifestMigration]]:
+    """
+    Discovers and returns a sorted list of all ManifestMigration subclasses available in the migrations package.
+    This function inspects the main migrations package and its submodules to find all classes that are subclasses of ManifestMigration,
+    excluding the ManifestMigration base class itself and any duplicates. The discovered migration classes are then sorted using the
+    _migration_order_key function to ensure they are returned in the correct order.
+
+    Returns:
+        List[Type[ManifestMigration]]: A list of discovered ManifestMigration subclasses, sorted by migration order.
+    """
+
     migration_classes = []
-    for name, obj in inspect.getmembers(sys.modules[migrations_pkg.__name__], inspect.isclass):
+    for _, obj in inspect.getmembers(sys.modules[migrations_pkg.__name__], inspect.isclass):
         if (
             issubclass(obj, ManifestMigration)
             and obj is not ManifestMigration
@@ -37,7 +66,7 @@ def _discover_migrations() -> List[Type[ManifestMigration]]:
     for _, module_name, _ in pkgutil.iter_modules(migrations_pkg.__path__):
         module = sys.modules.get(f"{migrations_pkg.__name__}.{module_name}")
         if module:
-            for name, obj in inspect.getmembers(module, inspect.isclass):
+            for _, obj in inspect.getmembers(module, inspect.isclass):
                 if (
                     issubclass(obj, ManifestMigration)
                     and obj is not ManifestMigration
@@ -47,7 +76,9 @@ def _discover_migrations() -> List[Type[ManifestMigration]]:
 
     # Sort by migration order key
     migration_classes.sort(key=_migration_order_key)
+
     return migration_classes
 
 
+# registered migrations
 MIGRATIONS: List[Type[ManifestMigration]] = _discover_migrations()