feat(db): add 'python -m app.db' CLI for migrations
D006: idempotent --migrate, --info (schema version + row counts), and --path. Used by deploy.sh to bring the DB up to date after a git pull without the web server running.
This commit is contained in:
1 parent
7f392ef908
commit
aef132fe94
1 file changed
+74
@@ -0,0 +1,74 @@
|
||||
#!/usr/bin/env python3
|
||||
"""CLI entry point for the SQLite store.
|
||||
|
||||
Usage:
|
||||
python -m app.db --migrate applies pending migrations (idempotent)
|
||||
python -m app.db --info prints schema version + table counts, exits
|
||||
python -m app.db --path prints the resolved DB path, exits
|
||||
|
||||
Used by ``deploy.sh`` (D004) to ensure the DB is up to date after a ``git pull``
|
||||
without needing the web server up.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
|
||||
from ..utils.constants import DATA_DB_PATH
|
||||
from .schema import SCHEMA_VERSION, apply_migrations
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="python -m app.db",
|
||||
description="Intrusion-detection SQLite store utilities.")
|
||||
parser.add_argument("--migrate", action="store_true",
|
||||
help="apply pending schema migrations (idempotent)")
|
||||
parser.add_argument("--info", action="store_true",
|
||||
help="print schema version + table/row counts")
|
||||
parser.add_argument("--path", action="store_true",
|
||||
help="print the resolved DB path and exit")
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
if args.path:
|
||||
print(DATA_DB_PATH)
|
||||
return 0
|
||||
|
||||
from . import get_connection
|
||||
|
||||
conn = get_connection()
|
||||
current = conn.execute("PRAGMA user_version").fetchone()[0]
|
||||
|
||||
if args.info:
|
||||
counts = {}
|
||||
tables = conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table' "
|
||||
"AND name NOT LIKE 'sqlite_%' ORDER BY name").fetchall()
|
||||
for t in tables:
|
||||
name = t["name"]
|
||||
try:
|
||||
counts[name] = conn.execute(
|
||||
f"SELECT COUNT(*) FROM \"{name}\"").fetchone()[0]
|
||||
except Exception:
|
||||
counts[name] = None
|
||||
print(f"db_path: {DATA_DB_PATH}")
|
||||
print(f"version: {current} (latest {SCHEMA_VERSION})")
|
||||
print("tables:")
|
||||
for name, n in counts.items():
|
||||
print(f" {name:<32} {n} rows")
|
||||
return 0
|
||||
|
||||
if args.migrate or current < SCHEMA_VERSION:
|
||||
before = current
|
||||
apply_migrations(conn)
|
||||
print(f"schema version {before} -> "
|
||||
f"{conn.execute('PRAGMA user_version').fetchone()[0]}")
|
||||
return 0
|
||||
|
||||
print(f"schema is already up to date (version {current})")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in new issue
Block a user