feat: add init option

Author: svazquezco

.pre-commit-config.yaml

@@ -35,6 +35,7 @@ repos:
         args: ["--config-file=pyproject.toml", "."]
         pass_filenames: false
         additional_dependencies:
-          - typer
-          - mpt_api_client
-          - pyairtable
+          - mpt_api_client==5.0.*
+          - typer>=0.9.0
+          - types-requests==2.32.*
+          - pyairtable==3.3.*

docs/PROJECT_DESCRIPTION.md

@@ -7,12 +7,16 @@ mpt-tool is a command-line utility to scaffold, run, and audit migrations for MP
     ```bash
       pip install mpt-tool
     ```
-2. **Create your first migration:**
+2. **Initialize the migration tool:**
+    ```bash
+      mpt-tool migrate --init
+    ```
+3. **Create your first migration:**
     ```bash
       mpt-tool migrate --new-data sync_users
     ```
-3. **Edit the generated file in the migrations/ folder**
-4. **Run all pending migrations**
+4. **Edit the generated file in the migrations/ folder**
+5. **Run all pending data migrations**
     ```bash
       mpt-tool migrate --data
     ```
@@ -32,7 +36,7 @@ Install with pip or your favorite PyPI package manager:
 ## Prerequisites
 
 - Python 3.12+ in your environment
-- A `migrations/` folder in your project (it will be created automatically the first time you create a migration)
+- A `migrations/` folder in your project (created automatically with `--init` or when you create your first migration)
 - Environment variables. See [Environment Variables](#environment-variables) for details.
 
 ## Environment Variables
@@ -80,6 +84,27 @@ Your Airtable table must have the following columns:
 2. Add the columns listed above with the specified field types
 3. Set the environment variables with your base ID and table name
 
+## Initialization
+
+Before using the migration tool for the first time, you should initialize it. This creates the necessary resources:
+
+```bash
+  mpt-tool migrate --init
+```
+
+This command creates:
+- The `migrations/` folder in your project root (if it doesn't exist)
+- The state storage:
+  - For **local storage**: creates `.migrations-state.json` file
+  - For **Airtable storage**: creates the table in Airtable with the required schema
+
+**When to use `--init`:**
+- First time setting up the tool in a project
+- When switching from local to Airtable storage (or vice versa)
+- When you need to recreate the state storage
+
+**Note:** If the state storage already exists, the command will fail with an error message. This prevents accidental data loss. If you need to reinitialize, manually delete the existing state file or table first.
+
 ## Usage
 
 ### Creating a New Migration
@@ -281,6 +306,14 @@ Run `mpt-tool --help` to see all available commands and params:
 
 ### Common Issues
 
+**Initialization fails - state already exists:**
+- Error: "Cannot initialize - State file already exists" (local storage) or similar for Airtable
+- Cause: The state storage has already been initialized
+- Solution: This is intentional to prevent data loss. If you need to reinitialize:
+  - For local storage: delete `.migrations-state.json` manually
+  - For Airtable: delete the table manually or use a different table name
+- Only reinitialize if you're certain you want to start fresh
+
 **Migrations not detected:**
 - Ensure files are in the `migrations/` folder
 - Verify filename follows the pattern: `<timestamp>_<migration_id>.py` (e.g., `20260121120000_migration_name.py`)

mpt_tool/cli.py

@@ -31,6 +31,9 @@ def migrate(  # noqa: WPS211
             metavar="MIGRATION_ID",
         ),
     ] = None,
+    init: Annotated[  # noqa: FBT002
+        bool, typer.Option("--init", help="Initialize migration tool resources.")
+    ] = False,
     new_data: Annotated[
         str | None,
         typer.Option(

mpt_tool/commands/factory.py

@@ -7,6 +7,7 @@
 from mpt_tool.commands.data import DataCommand
 from mpt_tool.commands.errors import CommandNotFoundError
 from mpt_tool.commands.fake import FakeCommand
+from mpt_tool.commands.init import InitCommand
 from mpt_tool.commands.list import ListCommand
 from mpt_tool.commands.new_data import NewDataCommand
 from mpt_tool.commands.new_schema import NewSchemaCommand
@@ -30,6 +31,8 @@ def get_instance(cls, param_data: dict[str, bool | str | None]) -> BaseCommand:
             CommandNotFoundError: If no command is found.
         """
         match param_data:  # noqa: WPS242
+            case {"init": True}:
+                return InitCommand()
             case {"check": True}:
                 return CheckCommand()
             case {"data": True}:

mpt_tool/commands/init.py

@@ -0,0 +1,22 @@
+from typing import override
+
+from mpt_tool.commands.base import BaseCommand
+from mpt_tool.use_cases import InitializeUseCase
+
+
+class InitCommand(BaseCommand):
+    """Initialize migration tool resources."""
+
+    @override
+    @property
+    def start_message(self) -> str:
+        return "Initializing migration tool..."
+
+    @override
+    @property
+    def success_message(self) -> str:
+        return "Migration tool initialized successfully."
+
+    @override
+    def run(self) -> None:
+        InitializeUseCase().execute()

mpt_tool/managers/errors.py

@@ -9,6 +9,10 @@ class CreateMigrationError(ManagerError):
     """Error creating the migration file."""
 
 
+class InitializationError(ManagerError):
+    """Error during initialization."""
+
+
 class InvalidStateError(ManagerError):
     """Error loading invalid state."""
 

mpt_tool/managers/file_migration.py

@@ -17,6 +17,11 @@ class FileMigrationManager:
 
     _migration_folder: Path = Path(MIGRATION_FOLDER)
 
+    @classmethod
+    def exists(cls) -> bool:
+        """Check if the file migration folder exists."""
+        return cls._migration_folder.exists()
+
     @classmethod
     def load_migration(cls, migration_file: MigrationFile) -> BaseMigration:
         """Loads a migration instance from a migration file.
@@ -61,7 +66,7 @@ def new_migration(cls, file_suffix: str, migration_type: MigrationTypeEnum) -> M
         Raises:
             CreateMigrationError: If an error occurs during migration creation.
         """
-        cls._migration_folder.mkdir(parents=True, exist_ok=True)
+        cls.new_migration_folder()
         try:
             migration_file = MigrationFile.new(migration_id=file_suffix, path=cls._migration_folder)
         except ValueError as error:
@@ -84,6 +89,11 @@ def new_migration(cls, file_suffix: str, migration_type: MigrationTypeEnum) -> M
         )
         return migration_file
 
+    @classmethod
+    def new_migration_folder(cls) -> None:
+        """Creates a new migration folder."""
+        cls._migration_folder.mkdir(parents=True, exist_ok=True)
+
     @classmethod
     def retrieve_migration_files(cls) -> tuple[MigrationFile, ...]:
         """Retrieves all migration files."""
@@ -99,7 +109,7 @@ def validate(cls) -> tuple[MigrationFile, ...]:
         Raises:
              MigrationFolderError: If an error occurs during migration validation.
         """
-        if not cls._migration_folder.exists():
+        if not cls.exists():
             raise MigrationFolderError(f"Migration folder not found: {cls._migration_folder}")
 
         migrations = cls._get_migration_files()

mpt_tool/managers/state/airtable.py

@@ -1,12 +1,14 @@
-from typing import override
+from typing import Any, override
 
+from pyairtable import Api
 from pyairtable.formulas import match
 from pyairtable.orm import Model, fields
+from requests import HTTPError
 
 from mpt_tool.config import get_airtable_config
 from mpt_tool.enums import MigrationTypeEnum
 from mpt_tool.managers import StateManager
-from mpt_tool.managers.errors import StateNotFoundError
+from mpt_tool.managers.errors import InitializationError, StateNotFoundError
 from mpt_tool.models import Migration
 
 
@@ -41,19 +43,13 @@ class AirtableStateManager(StateManager):
 
     @override
     @classmethod
-    def load(cls) -> dict[str, Migration]:
-        migrations = {}
-        for state in MigrationStateModel.all():
-            migration = Migration(
-                migration_id=state.migration_id,
-                order_id=state.order_id,
-                type=MigrationTypeEnum(state.type),
-                started_at=state.started_at,
-                applied_at=state.applied_at,
-            )
-            migrations[migration.migration_id] = migration
+    def exists(cls) -> bool:
+        try:
+            MigrationStateModel.meta.table.schema()
+        except HTTPError:
+            return False
 
-        return migrations
+        return True
 
     @override
     @classmethod
@@ -70,6 +66,67 @@ def get_by_id(cls, migration_id: str) -> Migration:
             applied_at=state.applied_at,
         )
 
+    @override
+    @classmethod
+    def initialize(cls) -> None:
+        api_key = get_airtable_config("api_key")
+        base_id = get_airtable_config("base_id")
+        table_name = get_airtable_config("table_name")
+        if not api_key or not base_id or not table_name:
+            raise InitializationError(
+                "Airtable configuration missing. Please set MPT_TOOL_STORAGE_AIRTABLE_API_KEY, "
+                "MPT_TOOL_STORAGE_AIRTABLE_BASE_ID, and MPT_TOOL_STORAGE_AIRTABLE_TABLE_NAME."
+            )
+
+        base = Api(api_key).base(base_id)
+        table_fields: list[dict[str, Any]] = [
+            {"name": "migration_id", "type": "singleLineText"},
+            {"name": "order_id", "type": "number", "options": {"precision": 0}},
+            {
+                "name": "type",
+                "type": "singleSelect",
+                "options": {"choices": [{"name": "data"}, {"name": "schema"}]},
+            },
+            {
+                "name": "started_at",
+                "type": "dateTime",
+                "options": {
+                    "dateFormat": {"name": "iso"},
+                    "timeFormat": {"name": "24hour"},
+                    "timeZone": "utc",
+                },
+            },
+            {
+                "name": "applied_at",
+                "type": "dateTime",
+                "options": {
+                    "dateFormat": {"name": "iso"},
+                    "timeFormat": {"name": "24hour"},
+                    "timeZone": "utc",
+                },
+            },
+        ]
+        try:
+            base.create_table(table_name, fields=table_fields)
+        except HTTPError as error:
+            raise InitializationError(str(error)) from error
+
+    @override
+    @classmethod
+    def load(cls) -> dict[str, Migration]:
+        migrations = {}
+        for state in MigrationStateModel.all():
+            migration = Migration(
+                migration_id=state.migration_id,
+                order_id=state.order_id,
+                type=MigrationTypeEnum(state.type),
+                started_at=state.started_at,
+                applied_at=state.applied_at,
+            )
+            migrations[migration.migration_id] = migration
+
+        return migrations
+
     @override
     @classmethod
     def new(cls, migration_id: str, migration_type: MigrationTypeEnum, order_id: int) -> Migration:

mpt_tool/managers/state/base.py

@@ -9,8 +9,8 @@ class StateManager(ABC):
 
     @classmethod
     @abstractmethod
-    def load(cls) -> dict[str, Migration]:
-        """Load migration states."""
+    def exists(cls) -> bool:
+        """Return True if storage exists."""
         raise NotImplementedError
 
     @classmethod
@@ -26,6 +26,22 @@ def get_by_id(cls, migration_id: str) -> Migration:
         """
         raise NotImplementedError
 
+    @classmethod
+    @abstractmethod
+    def initialize(cls) -> None:
+        """Initialize the state storage.
+
+        Raises:
+            InitializationError: If the storage already exists or cannot be created.
+        """
+        raise NotImplementedError
+
+    @classmethod
+    @abstractmethod
+    def load(cls) -> dict[str, Migration]:
+        """Load migration states."""
+        raise NotImplementedError
+
     @classmethod
     @abstractmethod
     def new(cls, migration_id: str, migration_type: MigrationTypeEnum, order_id: int) -> Migration: