zwitschi
aef132fe94
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.
75 lines
2.4 KiB
Python
75 lines
2.4 KiB
Python
#!/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())
|