feat: allow to run a single migration

Author: svazquezco

mpt_tool/cli.py

@@ -51,6 +51,9 @@ def migrate(  # noqa: WPS211
         ),
     ] = None,
     list: Annotated[bool, typer.Option("--list", help="List all migrations.")] = False,  # noqa: A002, FBT002
+    migration_id: Annotated[
+        str | None, typer.Argument(help="Optional migration ID for --data or --schema.")
+    ] = None,
 ) -> None:
     """Migrate command."""
     try:

mpt_tool/commands/data.py

@@ -8,6 +8,9 @@
 class DataCommand(BaseCommand):
     """Runs all data migrations."""
 
+    def __init__(self, migration_id: str | None = None) -> None:
+        self._migration_id = migration_id
+
     @override
     @property
     def start_message(self) -> str:
@@ -20,4 +23,4 @@ def success_message(self) -> str:
 
     @override
     def run(self) -> None:
-        RunMigrationsUseCase().execute(MigrationTypeEnum.DATA)
+        RunMigrationsUseCase().execute(MigrationTypeEnum.DATA, self._migration_id)

mpt_tool/commands/factory.py

@@ -35,10 +35,10 @@ def get_instance(cls, param_data: dict[str, bool | str | None]) -> BaseCommand:
                 return InitCommand()
             case {"check": True}:
                 return CheckCommand()
-            case {"data": True}:
-                return DataCommand()
-            case {"schema": True}:
-                return SchemaCommand()
+            case {"data": True, "migration_id": migration_id}:
+                return DataCommand(migration_id=cast(str | None, migration_id))
+            case {"schema": True, "migration_id": migration_id}:
+                return SchemaCommand(migration_id=cast(str | None, migration_id))
             case {"manual": manual_value} if manual_value is not None:
                 return ManualCommand(migration_id=cast(str, manual_value))
             case {"new_schema": new_schema_value} if new_schema_value is not None:

mpt_tool/commands/schema.py

@@ -8,6 +8,9 @@
 class SchemaCommand(BaseCommand):
     """Runs all schema migrations."""
 
+    def __init__(self, migration_id: str | None = None) -> None:
+        self._migration_id = migration_id
+
     @override
     @property
     def start_message(self) -> str:
@@ -20,4 +23,4 @@ def success_message(self) -> str:
 
     @override
     def run(self) -> None:
-        RunMigrationsUseCase().execute(MigrationTypeEnum.SCHEMA)
+        RunMigrationsUseCase().execute(MigrationTypeEnum.SCHEMA, self._migration_id)

mpt_tool/commands/validators.py

@@ -16,7 +16,14 @@ def validate(cls, command_params: dict[str, Any]) -> None:
         Raises:
             BadParameterError: When none or more than one param is used
         """
-        param_counts = sum(1 for param_value in command_params.values() if param_value)
+        migration_id = command_params.get("migration_id")
+        command_values = {
+            key: param_value for key, param_value in command_params.items() if key != "migration_id"
+        }
+        param_counts = sum(1 for param_value in command_values.values() if param_value)
+        if migration_id and not command_params.get("data") and not command_params.get("schema"):
+            raise BadParameterError("MIGRATION_ID can only be used with --data or --schema.")
+
         if not param_counts:
             raise BadParameterError("At least one param must be used.")
 

mpt_tool/use_cases/run_migrations.py

@@ -21,11 +21,14 @@ def __init__(
         self.file_migration_manager = file_migration_manager or FileMigrationManager()
         self.state_manager = state_manager or StateManagerFactory.get_instance()
 
-    def execute(self, migration_type: MigrationTypeEnum) -> None:  # noqa: WPS231
+    def execute(  # noqa: C901, WPS231
+        self, migration_type: MigrationTypeEnum, migration_id: str | None = None
+    ) -> None:
         """Run all migrations of a given type.
 
         Args:
             migration_type: The type of migrations to run.
+            migration_id: Optional migration id to run.
 
         Raises:
             RunMigrationError: If an error occurs during migration execution.
@@ -35,11 +38,17 @@ def execute(self, migration_type: MigrationTypeEnum) -> None:  # noqa: WPS231
         except MigrationFolderError as error:
             raise RunMigrationError(str(error)) from error
 
+        migration_files = self._filter_migration_files(migration_files, migration_id)
+
         for migration_file in migration_files:
             migration_instance = self._get_migration_instance_by_type(
                 migration_file, migration_type
             )
             if migration_instance is None:
+                if migration_id:
+                    raise RunMigrationError(
+                        f"Migration {migration_id} is not a {migration_type} migration"
+                    )
                 continue
 
             state = self._get_or_create_state(
@@ -63,6 +72,20 @@ def execute(self, migration_type: MigrationTypeEnum) -> None:  # noqa: WPS231
 
             self._save_state(state, status=MigrationStatusEnum.APPLIED)
 
+    def _filter_migration_files(
+        self, migration_files: tuple[MigrationFile, ...], migration_id: str | None
+    ) -> tuple[MigrationFile, ...]:
+        if migration_id is None:
+            return migration_files
+
+        filtered_migrations = tuple(
+            migration for migration in migration_files if migration.migration_id == migration_id
+        )
+        if not filtered_migrations:
+            raise RunMigrationError(f"Migration {migration_id} not found")
+
+        return filtered_migrations
+
     def _get_migration_instance_by_type(
         self, migration_file: MigrationFile, migration_type: MigrationTypeEnum
     ) -> BaseMigration | None:

tests/cli/test_cli.py

@@ -28,6 +28,14 @@ def test_migrate_command_multiple_params_error(runner):
     assert "Only one param can be used." in result.output
 
 
+def test_migrate_migration_id_without_type_error(runner):
+    result = runner.invoke(app, ["migrate", "fake_data_file_name"])
+
+    assert result.exit_code == 2, result.output
+    assert "Invalid value for migrate:" in result.output
+    assert "MIGRATION_ID can only be used with --data or" in result.output
+
+
 def test_migrate_data_duplicate_migration(runner, migration_folder):
     (migration_folder / "20250406020202_fake_file_name.py").touch()
     (migration_folder / "20260107010101_fake_file_name.py").touch()

tests/cli/test_cli_local_storage.py

@@ -51,6 +51,27 @@ def test_migrate_data_migration(migration_state_file, runner, log):
     assert "Migrations completed successfully." in result.output
 
 
+@freeze_time("2025-04-06 13:10:30")
+@pytest.mark.usefixtures("data_migration_file", "schema_migration_file")
+def test_migrate_data_single_migration(migration_state_file, runner, log):
+    result = runner.invoke(app, ["migrate", "--data", "fake_data_file_name"])
+
+    assert result.exit_code == 0, result.output
+    migration_state_data = json.loads(migration_state_file.read_text(encoding="utf-8"))
+    assert migration_state_data == {
+        "fake_data_file_name": {
+            "migration_id": "fake_data_file_name",
+            "order_id": 20250406020202,
+            "type": "data",
+            "started_at": "2025-04-06T13:10:30+00:00",
+            "applied_at": "2025-04-06T13:10:30+00:00",
+        }
+    }
+    assert "Running data migrations..." in result.output
+    assert "Running migration: fake_data_file_name" in log.text
+    assert "Migrations completed successfully." in result.output
+
+
 @freeze_time("2025-04-06 13:00:00")
 @pytest.mark.usefixtures("applied_migration")
 def test_migrate_skip_migration_already_applied(migration_state_file, runner, log):
@@ -90,6 +111,46 @@ def test_migrate_data_run_script_fail(migration_state_file, runner, log):
     assert "Migration fake_error_file_name failed: Fake Error" in result.output
 
 
+@pytest.mark.usefixtures("data_migration_file")
+def test_migrate_data_single_migration_not_found(runner):
+    result = runner.invoke(app, ["migrate", "--data", "not_existing_migration"])
+
+    assert result.exit_code == 1, result.output
+    assert "Error running data command: Migration not_existing_migration not found" in result.output
+
+
+@pytest.mark.usefixtures("data_migration_file", "schema_migration_file")
+def test_migrate_data_single_migration_wrong_type(runner):
+    result = runner.invoke(app, ["migrate", "--data", "fake_schema_file_name"])
+
+    assert result.exit_code == 1, result.output
+    assert (
+        "Error running data command: Migration fake_schema_file_name is not a data migration"
+        in result.output
+    )
+
+
+@freeze_time("2025-04-06 13:10:30")
+@pytest.mark.usefixtures("data_migration_file", "schema_migration_file")
+def test_migrate_schema_single_migration(migration_state_file, runner, log):
+    result = runner.invoke(app, ["migrate", "--schema", "fake_schema_file_name"])
+
+    assert result.exit_code == 0, result.output
+    migration_state_data = json.loads(migration_state_file.read_text(encoding="utf-8"))
+    assert migration_state_data == {
+        "fake_schema_file_name": {
+            "migration_id": "fake_schema_file_name",
+            "order_id": 20260101010101,
+            "type": "schema",
+            "started_at": "2025-04-06T13:10:30+00:00",
+            "applied_at": "2025-04-06T13:10:30+00:00",
+        }
+    }
+    assert "Running schema migrations..." in result.output
+    assert "Running migration: fake_schema_file_name" in log.text
+    assert "schema migrations applied successfully." in result.output
+
+
 @freeze_time("2025-04-06 10:11:24")
 @pytest.mark.usefixtures("data_migration_file")
 def test_migrate_manual(migration_state_file, runner):