CLI: enroll_db

README.md

@@ -109,6 +109,16 @@ 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
+   ```
+   This will create a new versioned database at the path, and copy the contents of the existing database into it. This is equivalent to calling `fastmigrate.enroll_db()`.
+
+   This command will offer to create an initial migration script for you, based on the current schema of the database. 
+
+
+
 
 ### How to enroll an existing, unversioned database into fastmigrate
 

enrolling.md

@@ -23,8 +23,8 @@ But if you are enrolling an existing db into fastmigrate, then you need to do th
 - Second, manually modify your current data to add fastmigrate version tag and set its version to 1. You can do this by using fastmigrate's internal API. Doing this constitutes asserting that the db is in fact in the state which would be produced by the migration script 0001. After doing this, fastmigrate will recognize your db as managed. Here is how to do it:
 
 ```python
-from fastmigrate.core import _ensure_meta_table, _set_db_version
-_ensure_meta_table("path/to/data.db")
+from fastmigrate.core import enroll_db, _set_db_version
+enroll_db("path/to/data.db")
 _set_db_version("path/to/data.db",1)
 ```
 

fastmigrate/cli.py

@@ -9,6 +9,10 @@
 from importlib.metadata import version
 
 from fastmigrate import core
+from rich.console import Console
+
+# Initialize Rich console
+console = Console()
 
 # Define constants - single source of truth for default values
 DEFAULT_DB = Path("data/database.db")
@@ -128,6 +132,57 @@ 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, # migrations dir, which may not exist
+    config_path:Path=DEFAULT_CONFIG, # Path to config file
+    force:bool=False # Force enrollment even if no migrations are available
+) -> None:
+    """Convert an unversioned SQLite database to a versioned one.
+
+    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 db_path.exists():
+        raise FileNotFoundError(f"Database file does not exist: {db_path}")    
+
+    initial_migration_path = migrations_path / "0001_initial.sql"    
+    if not force and not migrations_path.exists():
+        console.print(f"[red bold]No migrations directory or initial migration exists.[/red bold]")
+        while True:
+            answer = input(f"Want fastmigrate to generate the initial migration? [Y/n] ").strip().lower()
+            if answer.lower() in ('y', 'yes', ''):
+                migrations_path.mkdir(parents=True, exist_ok=True)
+                console.print(f"Created directory: {migrations_path}")
+                schema = core.get_db_schema(db_path) 
+                initial_migration_path.write_text(schema)
+                console.print(f"Created initial migration at '{initial_migration_path}'")
+                break
+            elif answer.lower() in ('n', 'no'):
+                answer2 = input(f"We [bold]strongly[/bold] recommend creating a migrations directory.\nWant to continue anyway? [N/y] ").strip().lower()
+                if answer2.lower() in ('n', 'no', ''):
+                    sys.exit(1)
+                elif answer2.lower() in ('y', 'yes'):
+                    console.print("[red]Continuing without a migrations directory or initial migration.[/red]")
+                    break
+            else:
+                console.print("Invalid input. Please enter 'y' or 'n'.") 
+
+    try:
+        success = core.enroll_db(db_path)
+    except core.MetaTableExists:
+        console.print(f"[red]Database at {db_path} is already versioned.[/red]")
+        sys.exit(1)
+    if not success:
+        console.print(f"[red]Failed to enroll the database at {db_path}.[/red]")
+        sys.exit(1)
+    console.print(f"[bold]Database at {db_path} is now versioned.[/bold]")
+
+
+
 @call_parse
 def run_migrations(
     db: Path = DEFAULT_DB, # Path to the SQLite database file
@@ -142,5 +197,22 @@ def run_migrations(
     db_path, migrations_path = _get_config(config_path, db, migrations)
     success = core.run_migrations(db_path, migrations_path, verbose=True)    
     if not success:
+        console.print(f"Ran migrations from {migrations_path} to {db_path}")
         sys.exit(1)
 
+
+@call_parse
+def get_db_schema(
+    db: Path = DEFAULT_DB, # Path to the SQLite database file
+    config_path: Path = DEFAULT_CONFIG # Path to config file
+) -> None:
+    """Get the schema of the SQLite database.
+
+    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)
+    schema = core.get_db_schema(db_path)
+    print(schema)
+    # if not success:
+    #     sys.exit(1)

fastmigrate/core.py

@@ -9,14 +9,15 @@
 import warnings
 from datetime import datetime
 from pathlib import Path
-from typing import Dict, Optional
+from typing import Dict, List, Optional, Tuple
 
 from rich.console import Console
 
 # Initialize Rich console
 console = Console()
 
-__all__ = ["run_migrations", "create_db", "get_db_version", "create_db_backup",
+__all__ = ["run_migrations", "create_db", "get_db_version",
+            "create_db_backup", "get_db_schema", "enroll_db", "MetaTableExists"
            # deprecated
            "ensure_versioned_db",
            "create_database_backup"]
@@ -36,7 +37,7 @@ def create_db(db_path:Path) -> int:
         db_path.parent.mkdir(parents=True, exist_ok=True)
         conn = sqlite3.connect(db_path)
         conn.close()
-        _ensure_meta_table(db_path)
+        enroll_db(db_path, err_if_versioned=False)
         return 0
     else:
         return get_db_version(db_path)
@@ -49,24 +50,41 @@ def ensure_versioned_db(db_path:Path) -> int:
                   stacklevel=2)
     return create_db(db_path)
 
-def _ensure_meta_table(db_path: Path) -> None:
+def _ensure_meta_table(db_path:Path) -> None:
+    "See enroll_db"
+    warnings.warn("_ensure_meta_table is deprecated, as it has been renamed to enroll_db, which is identical except adding a boolean return value",
+                 DeprecationWarning,
+                  stacklevel=2)
+    return enroll_db(db_path, err_if_versioned=False) 
+
+class MetaTableExists(Exception): pass   
+
+def enroll_db(db_path: Path, err_if_versioned:bool=True) -> bool:
     """Create the _meta table if it doesn't exist, with a single row constraint.
 
     Uses a single-row pattern with a PRIMARY KEY on a constant value (1).
     This ensures we can only have one row in the table.
 
+    Depending on how this function is called, it may raise an error if the
+    table already exists or return 0. This allows different levels of verbosity
+    to be used in different contexts.
+
     WARNING: users should call this directly only if preparing a
     versioned db manually, for instance, for testing or for enrolling
     a non-version db after verifying its values and schema are what
-    would be produced by migration scripts.
+    would be produced by migration scripts.    
 
     Args:
         db_path: Path to the SQLite database
+        err_if_versioned: If True, raises an error if the table already exists.
+                           If False, returns False if the table exists.
+
+    Returns:
+        bool: True if the table was created successfully, False otherwise
 
     Raises:
         FileNotFoundError: If database file doesn't exist
         sqlite3.Error: If unable to read or write to the database
-
     """
     db_path = Path(db_path)
     # First check if the file exists
@@ -100,12 +118,15 @@ def _ensure_meta_table(db_path: Path) -> None:
                     conn.execute("INSERT INTO _meta (id, version) VALUES (1, 0)")
             except sqlite3.Error as e:
                 raise sqlite3.Error(f"Failed to create _meta table: {e}")
+            return True
+        else:
+            if err_if_versioned: raise MetaTableExists("_meta table already exists")
+            else: return False
     except sqlite3.Error as e:
         raise sqlite3.Error(f"Failed to access database: {e}")
     finally:
         if conn:
-            conn.close()
-
+            conn.close()    
 
 def get_db_version(db_path: Path) -> int:
     """Get the current database version.
@@ -212,7 +233,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
 
 
@@ -360,6 +380,52 @@ def create_database_backup(db_path:Path) -> Path | None:
     return create_db_backup(db_path)
 
 
+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
+    """
+    db_path = Path(db_path)
+    # First check if the file exists
+    if not db_path.exists():
+        raise FileNotFoundError(f"Database file does not exist: {db_path}")
+
+    conn = None
+    try:
+        conn = sqlite3.connect(db_path)
+        sqls = []
+        # Get schema information for all objects
+        for row in conn.execute(
+            "select sql from sqlite_master where sql is not null"
+        ).fetchall():
+            sql = row[0]
+            if 'sqlite_stat1' in sql: continue
+            if 'sqlite_stat4' in sql: continue
+            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")
+            if not sql.strip().endswith(";"):
+                sql += ";"
+            sqls.append(sql)
+        return "\n".join(sqls)
+
+    except sqlite3.Error as e:
+        raise sqlite3.Error(f"Failed to retrieve database schema: {e}")
+    finally:
+        if conn:
+            conn.close()
+
+
 def execute_migration_script(db_path: Path, script_path: Path) -> bool:
     """Execute a migration script based on its file extension."""
     db_path = Path(db_path)

pyproject.toml

@@ -25,6 +25,8 @@ 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_get_db_schema = "fastmigrate.cli:get_db_schema"
 fastmigrate_run_migrations = "fastmigrate.cli:run_migrations"
 
 [tool.pytest]