feat: add --new-data and --new-schema options to migrate command

Author: svazquezco

.gitignore

@@ -168,6 +168,6 @@ cython_debug/
 .idea
 
 # Specific project gitignore
-migrations/
+migrations
 .migrations-state.json
 settings.py

.pre-commit-config.yaml

@@ -34,3 +34,5 @@ repos:
       - id: mypy
         args: ["--config-file=pyproject.toml", "."]
         pass_filenames: false
+        additional_dependencies:
+          - typer

Dockerfile

@@ -1,16 +1,17 @@
 FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim AS base
 
+COPY pyproject.toml uv.lock ./mpt_tool/
+
 WORKDIR /mpt_tool
 
-COPY pyproject.toml uv.lock ./
 RUN uv venv /opt/venv
 
 ENV VIRTUAL_ENV=/opt/venv
 ENV PATH=/opt/venv/bin:$PATH
 
 FROM base AS build
 
-COPY . /mpt_tool
+COPY . .
 
 RUN uv sync --frozen --no-cache --all-groups --active
 

compose.yaml

@@ -4,6 +4,5 @@ services:
     build:
       context: .
       target: dev
-    working_dir: /mpt_tool
     volumes:
       - .:/mpt_tool

mpt_tool/cli.py

@@ -1,48 +1,66 @@
+import datetime as dt
 from pathlib import Path
 
 import typer
 
-app = typer.Typer(help="MPT CLI - Migration tool for extensions.", no_args_is_help=True)
-
-
-@app.command()
-def init() -> None:
-    """Initialize the migrations tool workspace."""
-    root = Path.cwd()
-
-    settings_file = root / "settings.py"
-    if not settings_file.exists():
-        typer.secho("Settings file does not exist.", fg=typer.colors.RED)
-        raise typer.Exit(code=1)
-
-    migrations_dir = root / "migrations"
-    if migrations_dir.exists():
-        typer.secho("Migrations folder already exists.", fg=typer.colors.RED)
-        raise typer.Exit(code=1)
+from mpt_tool.constants import MIGRATION_FOLDER
+from mpt_tool.templates import MIGRATION_SCAFFOLDING_TEMPLATE
 
-    migrations_dir.mkdir(exist_ok=True)
-
-    init_file = migrations_dir / "__init__.py"
-    init_file.touch(exist_ok=True)
-
-    state_file = root / ".migrations-state.json"
-    if state_file.exists():
-        typer.secho("State file already exists.", fg=typer.colors.RED)
-        raise typer.Exit(code=1)
-
-    if not state_file.exists():
-        state_file.write_text("{}\n")
+app = typer.Typer(help="MPT CLI - Migration tool for extensions.", no_args_is_help=True)
 
 
 @app.callback()
 def callback() -> None:
     """MPT CLI - Migration tool for extensions."""
 
 
-@app.command()
-def migrate() -> None:
-    """Run the migration process."""
-    typer.echo("Hello World!")
+@app.command("migrate")
+def migrate(
+    new_data: str | None = typer.Option(  # noqa: WPS404
+        None,
+        "--new-data",
+        metavar="FILENAME",
+        help="Scaffold a new data migration script with the provided filename.",
+    ),
+    new_schema: str | None = typer.Option(  # noqa: WPS404
+        None,
+        "--new-schema",
+        metavar="FILENAME",
+        help="Scaffold a new schema migration script with the provided filename.",
+    ),
+) -> None:
+    """Migrate command."""
+    if new_data and new_schema:
+        raise typer.BadParameter(
+            "Options --new-data and --new-schema cannot be combined.",
+            param_hint="migrate",
+        )
+
+    if new_schema or new_data:
+        filename_suffix = new_data or new_schema
+        typer.echo(f"Scaffolding migration: {filename_suffix}.")
+        migration_folder = Path(MIGRATION_FOLDER)
+        migration_folder.mkdir(parents=True, exist_ok=True)
+        timestamp = dt.datetime.now(tz=dt.UTC).strftime("%Y%m%d%H%M%S")
+        # TODO: add filename validation
+        filename = f"{timestamp}_{filename_suffix}.py"
+        full_filename_path = migration_folder / filename
+        try:
+            full_filename_path.touch(exist_ok=False)
+        except FileExistsError:
+            typer.secho(f"File already exists: {filename}", fg=typer.colors.RED)
+            raise typer.Abort
+
+        full_filename_path.write_text(
+            encoding="utf-8",
+            data=MIGRATION_SCAFFOLDING_TEMPLATE.substitute(
+                command_name="DataBaseCommand" if new_data else "SchemaBaseCommand"
+            ),
+        )
+        typer.secho(f"Migration file: {filename} has been created.", fg=typer.colors.GREEN)
+        return
+
+    typer.secho("Running migrations is not implemented yet.", fg=typer.colors.YELLOW)
 
 
 def main() -> None:

mpt_tool/commands/__init__.py

@@ -0,0 +1,3 @@
+from mpt_tool.commands.base import DataBaseCommand, SchemaBaseCommand
+
+__all__ = ["DataBaseCommand", "SchemaBaseCommand"]

mpt_tool/commands/base.py

@@ -0,0 +1,24 @@
+from abc import ABC, abstractmethod
+
+from mpt_tool.commands.enums import MigrationTypeEnum
+
+
+class BaseCommand(ABC):
+    """Abstract base class for all migration commands."""
+
+    @abstractmethod
+    def run(self) -> None:
+        """Executes the command."""
+        raise NotImplementedError
+
+
+class DataBaseCommand(BaseCommand, ABC):
+    """Base command for data migrations."""
+
+    _type = MigrationTypeEnum.DATA
+
+
+class SchemaBaseCommand(BaseCommand, ABC):
+    """Base command for schema migrations."""
+
+    _type = MigrationTypeEnum.SCHEMA

mpt_tool/commands/enums.py

@@ -0,0 +1,13 @@
+from enum import StrEnum
+
+
+class MigrationTypeEnum(StrEnum):
+    """Enumeration of migration types.
+
+    Attributes:
+        DATA: Represents a data migration.
+        SCHEMA: Represents a schema migration.
+    """
+
+    DATA = "data"
+    SCHEMA = "schema"

mpt_tool/constants.py

@@ -0,0 +1 @@
+MIGRATION_FOLDER: str = "migrations"

mpt_tool/templates.py

@@ -0,0 +1,12 @@
+from string import Template
+
+MIGRATION_SCAFFOLDING_FILE_TEXT = """\
+from mpt_tool.commands import $command_name
+
+
+class Command($command_name):
+    def run(self):
+        # implement your logic here
+        pass
+"""
+MIGRATION_SCAFFOLDING_TEMPLATE: Template = Template(MIGRATION_SCAFFOLDING_FILE_TEXT)