✨ (pyproject.toml, main.py): add support for custom migrations path

Bump version to 0.2.0 to reflect new feature addition. Add a new option
to specify the path to the migrations folder, enhancing flexibility in
migration management.

Author: SpaceShaman

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "SQLift"
-version = "0.1.3"
+version = "0.2.0"
 description = "Simple CLI migration tool for SQL databases"
 license = "MIT"
 authors = ["SpaceShaman <[email protected]>"]

sqlift/main.py

@@ -2,7 +2,7 @@
 from typing import Optional
 
 from rich import print
-from typer import Argument, Typer
+from typer import Argument, Option, Typer
 from typing_extensions import Annotated
 
 from .clients import Client, get_client
@@ -11,12 +11,23 @@
 
 
 @app.command()
-def up(target_migration: Annotated[Optional[str], Argument()] = None) -> None:
+def up(
+    target_migration: Annotated[
+        Optional[str],
+        Argument(help="Name of the target migration to apply up to"),
+    ] = None,
+    migrations_path: Annotated[
+        str,
+        Option("--migrations-path", "-P", help="Path to migrations folder"),
+    ] = "migrations",
+) -> None:
     """Apply migrations up to the target migration or all if no target is provided."""
     client = get_client()
     _create_migrations_table_if_not_exists(client)
-    for migration_name in _get_migration_names(target_migration):
-        _apply_migration(client, migration_name)
+    for migration_name in _get_migration_names(
+        target_migration, migrations_path=migrations_path
+    ):
+        _apply_migration(client, migration_name, migrations_path=migrations_path)
     if target_migration:
         print(
             f"[bold green]All migrations up to {target_migration} applied successfully[/bold green] :thumbs_up:"
@@ -28,12 +39,21 @@ def up(target_migration: Annotated[Optional[str], Argument()] = None) -> None:
 
 
 @app.command()
-def down(target_migration: Annotated[Optional[str], Argument()] = None) -> None:
+def down(
+    target_migration: Annotated[
+        Optional[str], Argument(help="Name of the target migration to revert down to")
+    ] = None,
+    migrations_path: Annotated[
+        str, Option("--migrations-path", "-P", help="Path to migrations folder")
+    ] = "migrations",
+) -> None:
     """Revert migrations down to the target migration or all if no target is provided."""
     client = get_client()
     _create_migrations_table_if_not_exists(client)
-    for migration_name in _get_migration_names(target_migration, reverse=True):
-        _revert_migration(client, migration_name)
+    for migration_name in _get_migration_names(
+        target_migration, reverse=True, migrations_path=migrations_path
+    ):
+        _revert_migration(client, migration_name, migrations_path=migrations_path)
     if target_migration:
         print(
             f"[bold green]All migrations down to {target_migration} reverted successfully[/bold green] :thumbs_up:"
@@ -44,44 +64,50 @@ def down(target_migration: Annotated[Optional[str], Argument()] = None) -> None:
         )
 
 
-def _apply_migration(client: Client, migration_name: str) -> None:
+def _apply_migration(client: Client, migration_name: str, migrations_path: str) -> None:
     if _is_migration_recorded(client, migration_name):
         return
-    client.execute(_get_sql_up_command(migration_name))
+    client.execute(_get_sql_up_command(migration_name, migrations_path))
     _record_migration(client, migration_name)
     print(f"[green]- {migration_name}[/green] applied successfully")
 
 
-def _revert_migration(client: Client, migration_name: str) -> None:
+def _revert_migration(
+    client: Client, migration_name: str, migrations_path: str
+) -> None:
     if not _is_migration_recorded(client, migration_name):
         return
-    client.execute(_get_sql_down_command(migration_name))
+    client.execute(_get_sql_down_command(migration_name, migrations_path))
     _delete_migration_record(client, migration_name)
     print(f"[red]- {migration_name}[/red] reverted successfully")
 
 
 def _get_migration_names(
-    target_migration: str | None, reverse: bool = False
+    target_migration: str | None,
+    reverse: bool = False,
+    migrations_path: str = "migrations",
 ) -> list[str]:
     migration_names = sorted(
-        [migration_path.stem for migration_path in Path("migrations").glob("*.sql")],
+        [migration_path.stem for migration_path in Path(migrations_path).glob("*.sql")],
         reverse=reverse,
     )
     if target_migration:
         return migration_names[: migration_names.index(target_migration) + 1]
     return migration_names
 
 
-def _get_sql_commands(migration_name: str) -> list[str]:
-    return open(f"migrations/{migration_name}.sql").read().split("--DOWN")
+def _get_sql_commands(
+    migration_name: str, migrations_path: str = "migrations"
+) -> list[str]:
+    return open(Path(migrations_path) / f"{migration_name}.sql").read().split("--DOWN")
 
 
-def _get_sql_up_command(migration_name: str) -> str:
-    return _get_sql_commands(migration_name)[0]
+def _get_sql_up_command(migration_name: str, migrations_path: str) -> str:
+    return _get_sql_commands(migration_name, migrations_path)[0]
 
 
-def _get_sql_down_command(migration_name: str) -> str:
-    return _get_sql_commands(migration_name)[1]
+def _get_sql_down_command(migration_name: str, migrations_path: str) -> str:
+    return _get_sql_commands(migration_name, migrations_path)[1]
 
 
 def _record_migration(client: Client, migration_name: str) -> None: