updated the structure to address the versioning and the order of the migrations registered, add the metadata.applied_migrations object to be able to trace the applied MigrationTrace

Author: bazarnov

airbyte_cdk/manifest_migrations/README.md

@@ -6,44 +6,53 @@ This directory contains the logic and registry for manifest migrations in the Ai
 
 1. **Create a Migration File:**
    - Add a new Python file in the `migrations/` subdirectory.
-   - Name the file using the pattern: `<description>_v<major>_<minor>_<patch>__<order>.py`.
-     - Example: `http_requester_url_base_to_url_v6_45_2__0.py`
-   - The `<order>` integer is used to determine the order of migrations for the same version.
+   - Name the file using the pattern: `<description_of_the_migration>.py`.
+     - Example: `http_requester_url_base_to_url.py`
+   - The filename should be unique and descriptive.
 
 2. **Define the Migration Class:**
    - The migration class must inherit from `ManifestMigration`.
-   - Name the class using the pattern: `V_<major>_<minor>_<patch>_<Description>`.
-     - Example: `V_6_45_2_HttpRequesterUrlBaseToUrl`
+   - Name the class using a descriptive name (e.g., `HttpRequesterUrlBaseToUrl`).
    - Implement the following methods:
-     - `should_migrate(self, manifest: ManifestType) -> bool`: Return `True` if the migration should be applied to the given manifest.
-     - `migrate(self, manifest: ManifestType) -> None`: Perform the migration in-place.
-
-3. **Migration Versioning:**
-   - The migration version is extracted from the class name and used to determine applicability.
-   - Only manifests with a version less than or equal to the migration version will be migrated.
-
-4. **Component Type:**
-   - Use the `TYPE_TAG` constant to check the component type in your migration logic.
-
-5. **Examples:**
-   - See `migrations/http_requester_url_base_to_url_v6_45_2__0.py` and `migrations/http_requester_path_to_url_v6_45_2__1.py` for reference implementations.
-
-## Migration Registry
-
-- All migration classes in the `migrations/` folder are automatically discovered and registered in `migrations_registry.py`.
-- Migrations are applied in order, determined by the `<order>` suffix in the filename.
-
-## Testing
-
-- Ensure your migration is covered by unit tests.
-- Tests should verify both `should_migrate` and `migrate` behaviors.
+     - `should_migrate(self, manifest: ManifestType) -> bool`
+     - `migrate(self, manifest: ManifestType) -> None`
+     - `validate(self, manifest: ManifestType) -> bool`
+
+3. **Register the Migration:**
+   - Open `migrations/registry.yaml`.
+   - Add an entry under the appropriate version, or create a new version section if needed.
+   - Each migration entry should include:
+     - `name`: The filename (without `.py`)
+     - `order`: The order in which this migration should be applied for the version
+     - `description`: A short description of the migration
+
+   Example:
+   ```yaml
+   manifest_migrations:
+     - version: 6.45.2
+       migrations:
+         - name: http_requester_url_base_to_url
+           order: 1
+           description: |
+             This migration updates the `url_base` field in the `HttpRequester` component spec to `url`.
+   ```
+
+4. **Testing:**
+   - Ensure your migration is covered by unit tests.
+   - Tests should verify both `should_migrate`, `migrate`, and `validate` behaviors.
+
+## Migration Discovery
+
+- Migrations are discovered and registered automatically based on the entries in `migrations/registry.yaml`.
+- Do not modify the migration registry in code manually.
+- If you need to skip certain component types, use the `NON_MIGRATABLE_TYPES` list in `manifest_migration.py`.
 
 ## Example Migration Skeleton
 
 ```python
 from airbyte_cdk.manifest_migrations.manifest_migration import TYPE_TAG, ManifestMigration, ManifestType
 
-class V_1_2_3_Example(ManifestMigration):
+class ExampleMigration(ManifestMigration):
     component_type = "ExampleComponent"
     original_key = "old_key"
     replacement_key = "new_key"
@@ -54,12 +63,10 @@ class V_1_2_3_Example(ManifestMigration):
     def migrate(self, manifest: ManifestType) -> None:
         manifest[self.replacement_key] = manifest[self.original_key]
         manifest.pop(self.original_key, None)
-```
 
-## Additional Notes
-
-- Do not modify the migration registry manually; it will pick up all valid migration classes automatically.
-- If you need to skip certain component types, use the `NON_MIGRATABLE_TYPES` list in `manifest_migration.py`.
+    def validate(self, manifest: ManifestType) -> bool:
+        return self.replacement_key in manifest and self.original_key not in manifest
+```
 
 ---
 

airbyte_cdk/manifest_migrations/__init__.py

@@ -0,0 +1,3 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#

airbyte_cdk/manifest_migrations/exceptions.py

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
 #
 
 

airbyte_cdk/manifest_migrations/manifest_migration.py

@@ -1,29 +1,57 @@
-# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
 
-import re
-from abc import abstractmethod
-from typing import Any, Dict
 
-from packaging.version import Version
+from abc import ABC, abstractmethod
+from dataclasses import asdict, dataclass
+from typing import Any, Dict
 
 ManifestType = Dict[str, Any]
 
 
 TYPE_TAG = "type"
 
 NON_MIGRATABLE_TYPES = [
+    # more info here: https://github.com/airbytehq/airbyte-internal-issues/issues/12423
     "DynamicDeclarativeStream",
 ]
 
 
-class ManifestMigration:
+@dataclass
+class MigrationTrace:
+    """
+    This class represents a migration that has been applied to the manifest.
+    It contains information about the migration, including the version it was applied from,
+    the version it was applied to, and the time it was applied.
+    """
+
+    from_version: str
+    to_version: str
+    migration: str
+    migrated_at: str
+
+    def as_dict(self) -> dict:
+        return asdict(self)
+
+
+class ManifestMigration(ABC):
+    """
+    Base class for manifest migrations.
+    This class provides a framework for migrating manifest components.
+    It defines the structure for migration classes, including methods for checking if a migration is needed,
+    performing the migration, and validating the migration.
+    """
+
+    def __init__(self) -> None:
+        self.is_migrated: bool = False
+
     @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.
         """
@@ -34,17 +62,15 @@ 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
         """
 
-    @property
-    def migration_version(self) -> Version:
+    @abstractmethod
+    def validate(self, manifest: ManifestType) -> bool:
         """
-        Get the migration version.
+        Validate the manifest to ensure the migration was successfully applied.
 
-        :return: The migration version as a Version object
+        :param manifest: The manifest to validate
         """
-        return Version(self._get_migration_version())
 
     def _is_component(self, obj: Dict[str, Any]) -> bool:
         """
@@ -95,6 +121,8 @@ def _process_manifest(self, obj: Any) -> None:
                 if self.should_migrate(obj):
                     # Perform the migration, if needed
                     self.migrate(obj)
+                    # validate the migration
+                    self.is_migrated = self.validate(obj)
 
             # Process all values in the dictionary
             for value in list(obj.values()):
@@ -104,24 +132,3 @@ def _process_manifest(self, obj: Any) -> None:
             # Process all items in the list
             for item in obj:
                 self._process_manifest(item)
-
-    def _get_migration_version(self) -> str:
-        """
-        Get the migration version from the class name.
-        The migration version is extracted from the class name using a regular expression.
-        The expected format is "V_<major>_<minor>_<patch>_<migration_name>".
-
-        For example, "V_6_45_2_HttpRequesterPathToUrl" -> "6.45.2"
-
-        :return: The migration version as a string in the format "major.minor.patch"
-        :raises ValueError: If the class name does not match the expected format
-        """
-
-        class_name = self.__class__.__name__
-        migration_version = re.search(r"V_(\d+_\d+_\d+)", class_name)
-        if migration_version:
-            return migration_version.group(1).replace("_", ".")
-        else:
-            raise ValueError(
-                f"Invalid migration class name, make sure the class name has the version (e.g `V_<major>_<minor>_<patch>_<migration_name>`): {class_name}"
-            )

airbyte_cdk/manifest_migrations/migration_handler.py

@@ -1,9 +1,11 @@
 #
-# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
 #
 
+
 import copy
 import logging
+from datetime import datetime, timezone
 from typing import Type
 
 from packaging.version import Version
@@ -14,11 +16,16 @@
 from airbyte_cdk.manifest_migrations.manifest_migration import (
     ManifestMigration,
     ManifestType,
+    MigrationTrace,
 )
 from airbyte_cdk.manifest_migrations.migrations_registry import (
-    MIGRATIONS,
+    MANIFEST_MIGRATIONS,
 )
 
+METADATA_TAG = "metadata"
+MANIFEST_VERSION_TAG = "version"
+APPLIED_MIGRATIONS_TAG = "applied_migrations"
+
 LOGGER = logging.getLogger("airbyte.cdk.manifest_migrations")
 
 
@@ -30,7 +37,6 @@ 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:
         """
@@ -45,14 +51,21 @@ def apply_migrations(self) -> ManifestType:
                           manifest if any migration failed.
         """
         try:
-            for migration_cls in MIGRATIONS:
-                self._handle_migration(migration_cls)
+            manifest_version = self._get_manifest_version()
+            for migration_version, migrations in MANIFEST_MIGRATIONS.items():
+                for migration_cls in migrations:
+                    self._handle_migration(migration_cls, manifest_version, migration_version)
             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:
+    def _handle_migration(
+        self,
+        migration_class: Type[ManifestMigration],
+        manifest_version: str,
+        migration_version: str,
+    ) -> None:
         """
         Handles a single manifest migration by instantiating the migration class and processing the manifest.
 
@@ -64,21 +77,67 @@ def _handle_migration(self, migration_class: Type[ManifestMigration]) -> None:
         """
         try:
             migration_instance = migration_class()
-            # check if the migration is supported for the given manifest version
-            if self._manifest_version <= migration_instance.migration_version:
+            if self._version_is_valid_for_migration(manifest_version, migration_version):
                 migration_instance._process_manifest(self._migrated_manifest)
+                if migration_instance.is_migrated:
+                    # set the updated manifest version, after migration has been applied
+                    self._set_manifest_version(migration_version)
+                    # set the migration trace
+                    self._set_migration_trace(migration_class, manifest_version, migration_version)
             else:
                 LOGGER.info(
-                    f"Manifest migration: `{migration_class.__name__}` is not supported for the given manifest version `{self._manifest_version}`.",
+                    f"Manifest migration: `{migration_instance.__name__}` is not supported for the given manifest version `{manifest_version}`.",
                 )
         except Exception as e:
             raise ManifestMigrationException(str(e)) from e
 
-    def _get_manifest_version(self) -> Version:
+    def _get_manifest_version(self) -> str:
         """
         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")))
+        return str(self._migrated_manifest.get(MANIFEST_VERSION_TAG, "0.0.0"))
+
+    def _version_is_valid_for_migration(
+        self, manifest_version: str, migration_version: str
+    ) -> bool:
+        return Version(manifest_version) <= Version(migration_version)
+
+    def _set_manifest_version(self, version: str) -> None:
+        """
+        Set the manifest version in the manifest.
+
+        :param version: The version to set
+        """
+        self._migrated_manifest[MANIFEST_VERSION_TAG] = version
+
+    def _set_migration_trace(
+        self,
+        migration_instance: Type[ManifestMigration],
+        manifest_version: str,
+        migration_version: str,
+    ) -> None:
+        """
+        Set the migration trace in the manifest.
+
+        :param migration_instance: The migration instance to set
+        :param manifest_version: The manifest version before migration
+        :param migration_version: The manifest version after migration
+        """
+
+        if METADATA_TAG not in self._migrated_manifest:
+            self._migrated_manifest[METADATA_TAG] = {}
+        if APPLIED_MIGRATIONS_TAG not in self._migrated_manifest[METADATA_TAG]:
+            self._migrated_manifest[METADATA_TAG][APPLIED_MIGRATIONS_TAG] = []
+
+        migration_trace = MigrationTrace(
+            from_version=manifest_version,
+            to_version=migration_version,
+            migration=migration_instance.__name__,
+            migrated_at=datetime.now(tz=timezone.utc).isoformat(),
+        ).as_dict()
+
+        if migration_version not in self._migrated_manifest[METADATA_TAG][APPLIED_MIGRATIONS_TAG]:
+            self._migrated_manifest[METADATA_TAG][APPLIED_MIGRATIONS_TAG].append(migration_trace)

airbyte_cdk/manifest_migrations/migrations/__init__.py

@@ -0,0 +1,4 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+