CLI: enroll_db

README.md

@@ -109,6 +109,11 @@ Here are some commands:
    ```
    Run all needed migrations on the db. Fails if a migration fails, or if there is no managed db at the path. This is equivalent to calling `fastmigrate.run_migrations()`
 
+5. **Enroll an existing db**:
+   ```
+   fastmigrate_enroll_db --db path/to/data.db
+   ```
+   Enroll an existing SQLite database for versioning, adding a default initial migration called `0001-initial.sql`, then running it. Running the initial migration will set the version to 1. This is equivalent to calling `fastmigrate.enroll_db()`
 
 ### How to enroll an existing, unversioned database into fastmigrate
 

fastmigrate/cli.py

@@ -128,6 +128,28 @@ def create_db(
         print(f"Unexpected error: {e}")
         sys.exit(1)
 
+@call_parse
+def enroll_db(
+    db: Path = DEFAULT_DB, # Path to the SQLite database file
+    migrations: Path = DEFAULT_MIGRATIONS, # Path to the migrations directory
+    config_path: Path = DEFAULT_CONFIG # Path to config file
+) -> None:
+    """Enroll an existing SQLite database for versioning, adding a default initial migration, then running it.
+
+    Note: command line arguments take precedence over values from a
+    config file, unless they are equal to default values.
+    """
+    db_path, migrations_path = _get_config(config_path, db, migrations)
+    if not migrations_path.exists(): migrations_path.mkdir(parents=True)
+    initial_migration = migrations_path / "0001-initial.sql"
+    schema = core.get_db_schema(db_path)    
+    initial_migration.write_text(schema)    
+    core._ensure_meta_table(db_path)
+    success = core.run_migrations(db_path, migrations_path, verbose=True)    
+    if not success:
+        sys.exit(1)
+
+
 @call_parse
 def run_migrations(
     db: Path = DEFAULT_DB, # Path to the SQLite database file

fastmigrate/core.py

@@ -10,6 +10,9 @@
 from datetime import datetime
 from pathlib import Path
 from typing import Dict, Optional
+from io import StringIO
+from apsw import Connection
+from apsw.shell import Shell
 
 from rich.console import Console
 
@@ -98,6 +101,7 @@ def _ensure_meta_table(db_path: Path) -> None:
                         """
                     )
                     conn.execute("INSERT INTO _meta (id, version) VALUES (1, 0)")
+                    print('Database is enrolled')
             except sqlite3.Error as e:
                 raise sqlite3.Error(f"Failed to create _meta table: {e}")
     except sqlite3.Error as e:
@@ -199,7 +203,6 @@ def get_migration_scripts(migrations_dir: Path) -> Dict[int, Path]:
     """
     migrations_dir = Path(migrations_dir)
     migration_scripts: Dict[int, Path] = {}
-
     if not migrations_dir.exists():
         return migration_scripts
 
@@ -212,7 +215,6 @@ def get_migration_scripts(migrations_dir: Path) -> Dict[int, Path]:
                     f"{migration_scripts[version]} and {file_path}"
                 )
             migration_scripts[version] = file_path
-
     return migration_scripts
 
 
@@ -507,3 +509,30 @@ def run_migrations(
     except Exception as e:
         console.print(f"[bold red]Error:[/bold red] {e}")
         return False
+
+def get_db_schema(db_path: Path) -> str:
+    """Get the SQL schema of a SQLite database file.
+
+    This function retrieves the CREATE statements for all tables, 
+    indices, triggers, and views in the database.
+
+    Args:
+        db_path: Path to the SQLite database file
+
+    Returns:
+        str: The SQL schema as a string
+
+    Raises:
+        FileNotFoundError: If database file doesn't exist
+        sqlite3.Error: If unable to access the database
+    """
+
+    if not Path(db_path).exists(): raise FileNotFoundError(f"Database does not exist: {db_path}")
+    conn = Connection(str(db_path))
+    out = StringIO()
+    shell = Shell(stdout=out, db=conn)
+    shell.process_command('.schema')
+    sql = out.getvalue()
+    if 'CREATE TABLE' in sql: sql = sql.replace("CREATE TABLE", "CREATE TABLE IF NOT EXISTS")
+    if 'CREATE INDEX' in sql: sql = sql.replace("CREATE INDEX", "CREATE INDEX IF NOT EXISTS")    
+    return sql

pyproject.toml

@@ -25,6 +25,7 @@ dev = [
 fastmigrate_backup_db = "fastmigrate.cli:backup_db"
 fastmigrate_check_version = "fastmigrate.cli:check_version"
 fastmigrate_create_db = "fastmigrate.cli:create_db"
+fastmigrate_enroll_db = "fastmigrate.cli:enroll_db"
 fastmigrate_run_migrations = "fastmigrate.cli:run_migrations"
 
 [tool.pytest]