refactor: add RunMigrationsUseCase

Author: svazquezco

mpt_tool/cli.py

@@ -1,14 +1,15 @@
 import datetime as dt
+import logging
 from pathlib import Path
 from typing import Annotated
 
 import typer
 
 from mpt_tool.constants import MIGRATION_FOLDER
 from mpt_tool.enums import MigrationTypeEnum
-from mpt_tool.errors import StateNotFoundError
-from mpt_tool.managers import FileMigrationManager, FileStateManager, MigrationFolderError
+from mpt_tool.errors import RunMigrationError
 from mpt_tool.templates import MIGRATION_SCAFFOLDING_TEMPLATE
+from mpt_tool.use_cases import RunMigrationsUseCase
 
 app = typer.Typer(help="MPT CLI - Migration tool for extensions.", no_args_is_help=True)
 
@@ -50,37 +51,13 @@ def migrate(  # noqa: C901, WPS238, WPS210, WPS213, WPS231
         migration_type = MigrationTypeEnum.DATA if data else MigrationTypeEnum.SCHEMA
         typer.echo(f"Running {migration_type} migrations...")
 
+        run_migration = RunMigrationsUseCase()
         try:
-            validated_migration_files = FileMigrationManager.validate()
-        except MigrationFolderError as error:
-            typer.secho(str(error), fg=typer.colors.RED)
+            run_migration.execute(migration_type)
+        except RunMigrationError as error:
+            typer.secho(f"Error running migrations: {error!s}", fg=typer.colors.RED)
             raise typer.Abort
 
-        for migration_file in validated_migration_files:
-            migration_id = migration_file.migration_id
-            try:
-                state = FileStateManager.get_by_id(migration_id)
-            except StateNotFoundError:
-                state = FileStateManager.new(migration_id, migration_type, migration_file.order_id)
-
-            if state.started_at is not None:
-                continue
-
-            typer.secho(f"Running migration: {migration_id}...", fg=typer.colors.GREEN)
-            state.start()
-            migration_module = FileMigrationManager.load_migration(migration_file)
-            migration_instance = migration_module.Command()
-            try:
-                migration_instance.run()
-            except Exception as error:
-                state.failed()
-                FileStateManager.save_state(state)
-                typer.secho(f"Migration {migration_id} failed: {error}", fg=typer.colors.RED)
-                raise typer.Abort
-
-            state.applied()
-            FileStateManager.save_state(state)
-
         typer.secho("Migrations completed successfully.", fg=typer.colors.GREEN)
         return
 
@@ -89,7 +66,6 @@ def migrate(  # noqa: C901, WPS238, WPS210, WPS213, WPS231
         typer.echo(f"Scaffolding migration: {filename_suffix}.")
         migration_folder = Path(MIGRATION_FOLDER)
         migration_folder.mkdir(parents=True, exist_ok=True)
-        # TODO: add init file
         timestamp = dt.datetime.now(tz=dt.UTC).strftime("%Y%m%d%H%M%S")
         # TODO: add filename validation
         filename = f"{timestamp}_{filename_suffix}.py"
@@ -111,4 +87,5 @@ def migrate(  # noqa: C901, WPS238, WPS210, WPS213, WPS231
 
 def main() -> None:
     """Entry point for the CLI."""
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
     app()

mpt_tool/commands/base.py

@@ -1,14 +1,23 @@
 import logging
 from abc import ABC, abstractmethod
 
+from mpt_tool.enums import MigrationTypeEnum
+
 logger = logging.getLogger(__name__)
 
 
 class BaseCommand(ABC):
     """Abstract base class for all migration commands."""
 
+    _type: MigrationTypeEnum
+
     def __init__(self):
-        self.logger = logger
+        self.log = logger
+
+    @property
+    def type(self) -> MigrationTypeEnum:
+        """The type of migration this command represents."""
+        return self._type
 
     @abstractmethod
     def run(self) -> None:

mpt_tool/errors.py

@@ -17,5 +17,9 @@ class MigrationFolderError(BaseError):
     """Error accessing migrations folder."""
 
 
+class RunMigrationError(BaseError):
+    """Error running migrations."""
+
+
 class StateNotFoundError(BaseError):
     """Error getting state from state file."""

mpt_tool/managers.py

@@ -16,7 +16,6 @@ class FileMigrationManager:
     """Manages migration files."""
 
     _migration_folder: Path = Path(MIGRATION_FOLDER)
-    _validated_migrations: tuple[MigrationFile, ...] = ()
 
     @classmethod
     def validate(cls) -> tuple[MigrationFile, ...]:
@@ -25,9 +24,14 @@ def validate(cls) -> tuple[MigrationFile, ...]:
             raise MigrationFolderError(f"Migration folder not found: {cls._migration_folder}")
 
         migrations = tuple(
-            MigrationFile.build_from_path(path)
-            for path in cls._migration_folder.glob("*.py")
-            if re.match(r"\d+_.*\.py", path.name)
+            sorted(
+                (
+                    MigrationFile.build_from_path(path)
+                    for path in cls._migration_folder.glob("*.py")
+                    if re.match(r"\d+_.*\.py", path.name)
+                ),
+                key=lambda migration_file: migration_file.order_id,
+            )
         )
         if not migrations:
             raise MigrationFolderError(f"No migration files found in {cls._migration_folder}")
@@ -39,8 +43,6 @@ def validate(cls) -> tuple[MigrationFile, ...]:
                 f"Duplicate migration filename found: {duplicated_migrations[0]}"
             )
 
-        cls._validated_migrations = migrations
-
         return migrations
 
     @classmethod

mpt_tool/use_cases/__init__.py

@@ -0,0 +1,3 @@
+from mpt_tool.use_cases.run_migrations import RunMigrationsUseCase
+
+__all__ = ["RunMigrationsUseCase"]

mpt_tool/use_cases/run_migrations.py

@@ -0,0 +1,80 @@
+import logging
+
+from mpt_tool.enums import MigrationTypeEnum
+from mpt_tool.errors import (
+    LoadMigrationError,
+    MigrationFolderError,
+    RunMigrationError,
+    StateNotFoundError,
+)
+from mpt_tool.managers import FileMigrationManager, FileStateManager
+from mpt_tool.models import Migration
+
+logger = logging.getLogger(__name__)
+
+
+class RunMigrationsUseCase:
+    """Service to run migrations."""
+
+    def __init__(
+        self,
+        file_migration_manager: FileMigrationManager | None = None,
+        state_manager: FileStateManager | None = None,
+    ):
+        self.file_migration_manager = file_migration_manager or FileMigrationManager
+        self.state_manager = state_manager or FileStateManager
+
+    def execute(self, migration_type: MigrationTypeEnum) -> None:  # noqa: C901, WPS213, WPS231
+        """Run all migrations of a given type.
+
+        Args:
+            migration_type: The type of migrations to run.
+
+        Raises:
+            RunMigrationError: If an error occurs during migration execution.
+        """
+        try:
+            migration_files = self.file_migration_manager.validate()
+        except MigrationFolderError as error:
+            raise RunMigrationError(str(error)) from error
+
+        for migration_file in migration_files:
+            try:
+                migration = self.file_migration_manager.load_migration(migration_file)
+            except LoadMigrationError as error:
+                raise RunMigrationError(str(error)) from error
+
+            migration_instance = migration.Command()
+            if migration_instance.type != migration_type:
+                continue
+
+            state = self._get_or_create_state(
+                migration_file.migration_id, migration_type, migration_file.order_id
+            )
+            if state.applied_at is not None:
+                logger.debug("Skipping applied migration: %s", migration_file.migration_id)
+                continue
+
+            logger.info("Running migration: %s", migration_file.migration_id)
+            state.start()
+            try:
+                migration_instance.run()
+            except Exception as error:
+                state.failed()
+                self.state_manager.save_state(state)
+                raise RunMigrationError(
+                    f"Migration {migration_file.migration_id} failed: {error!s}"
+                ) from error
+
+            state.applied()
+            self.state_manager.save_state(state)
+
+    def _get_or_create_state(
+        self, migration_id: str, migration_type: MigrationTypeEnum, order_id: int
+    ) -> Migration:
+        try:
+            state = self.state_manager.get_by_id(migration_id)
+        except StateNotFoundError:
+            state = self.state_manager.new(migration_id, migration_type, order_id)
+
+        return state

pyproject.toml

@@ -87,7 +87,6 @@ show-source = true
 statistics = false
 per-file-ignores = [
   "tests/*: WPS202",
-  "mpt_tool/cli.py: WPS232",
 ]
 
 [tool.ruff]

tests/conftest.py

@@ -0,0 +1,9 @@
+import logging
+
+import pytest
+
+
+@pytest.fixture
+def log(caplog):
+    caplog.set_level(logging.DEBUG)
+    return caplog

tests/test_cli.py

@@ -18,13 +18,12 @@ def mock_chdir(monkeypatch, tmp_path):
 def migration_folder(tmp_path):
     migration_folder_path = tmp_path / MIGRATION_FOLDER
     migration_folder_path.mkdir(exist_ok=True)
-    (migration_folder_path / "__init__.py").touch()
     return migration_folder_path
 
 
 @pytest.fixture
 def data_migration_file(migration_folder):
-    migration_id = "fake_file_name"
+    migration_id = "fake_data_file_name"
     migration_file = migration_folder / f"20250406020202_{migration_id}.py"
     migration_file.write_text(
         encoding="utf-8",
@@ -40,9 +39,17 @@ def data_migration_file_error(migration_folder):
     file_data = MIGRATION_SCAFFOLDING_TEMPLATE.substitute(command_name="DataBaseCommand").replace(
         "pass", "raise Exception('Fake Error')"
     )
+    migration_file.write_text(encoding="utf-8", data=file_data)
+    return {"migration_id": migration_id, "full_filename": migration_file}
+
+
+@pytest.fixture
+def schema_migration_file(migration_folder):
+    migration_id = "fake_schema_file_name"
+    migration_file = migration_folder / f"20260101010101_{migration_id}.py"
     migration_file.write_text(
         encoding="utf-8",
-        data=file_data,
+        data=MIGRATION_SCAFFOLDING_TEMPLATE.substitute(command_name="SchemaBaseCommand"),
     )
     return {"migration_id": migration_id, "full_filename": migration_file}
 
@@ -84,20 +91,57 @@ def test_migrate_command_multiple_options_error(runner):
 
 
 @freeze_time("2025-04-06 13:00:00")
-def test_migrate_data_migration(data_migration_file, migration_state_file, runner):
+def test_migrate_data_migration(
+    data_migration_file, schema_migration_file, migration_state_file, runner, log
+):
     result = runner.invoke(app, ["migrate", "--data"])
 
     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_file_name"] == {
-        "migration_id": "fake_file_name",
-        "order_id": 20250406020202,
-        "type": "data",
-        "started_at": "2025-04-06T13:00:00+00:00",
-        "applied_at": "2025-04-06T13:00:00+00:00",
+    assert migration_state_data == {
+        "fake_data_file_name": {
+            "migration_id": "fake_data_file_name",
+            "order_id": 20250406020202,
+            "type": "data",
+            "started_at": "2025-04-06T13:00:00+00:00",
+            "applied_at": "2025-04-06T13:00:00+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")
+def test_migrate_skip_migration_already_applied(
+    data_migration_file, migration_state_file, runner, log
+):
+    applied_state_data = {
+        "fake_data_file_name": {
+            "migration_id": "fake_data_file_name",
+            "order_id": 20250406020202,
+            "type": "data",
+            "started_at": "2025-04-06T13:00:00+00:00",
+            "applied_at": "2025-04-06T13:00:00+00:00",
+        }
+    }
+    migration_state_file.write_text(data=json.dumps(applied_state_data))
+
+    result = runner.invoke(app, ["migrate", "--data"])
+
+    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:00:00+00:00",
+            "applied_at": "2025-04-06T13:00:00+00:00",
+        }
     }
     assert "Running data migrations..." in result.output
-    assert "Running migration: fake_file_name" in result.output
+    assert "Skipping applied migration: fake_data_file_name" in log.text
     assert "Migrations completed successfully." in result.output
 
 
@@ -119,7 +163,7 @@ def test_migrate_data_duplicate_migration(runner, migration_folder):
 
 
 @freeze_time("2025-04-06 13:00:00")
-def test_migrate_data_run_script_fail(data_migration_file_error, migration_state_file, runner):
+def test_migrate_data_run_script_fail(data_migration_file_error, migration_state_file, runner, log):
     result = runner.invoke(app, ["migrate", "--data"])
 
     assert result.exit_code == 1, result.output
@@ -132,7 +176,7 @@ def test_migrate_data_run_script_fail(data_migration_file_error, migration_state
         "applied_at": None,
     }
     assert "Running data migrations..." in result.output
-    assert "Running migration: fake_error_file_name" in result.output
+    assert "Running migration: fake_error_file_name" in log.text
     assert "Migration fake_error_file_name failed: Fake Error" in result.output