""" Provision the TSAPI series database schema from the canonical SQL files. This is the SINGLE SOURCE OF TRUTH provisioner for the TSAPI timeseries PostgreSQL schema. Both the Python library (TSAPI_SERIESDB_PG_PY) and the C++ library (TSAPI_POSTGRES_CPP) target the schema it creates. Do not fork the SQL into either project -- edit it here. It executes the numbered DDL files in ./sql/ in order: 00_extensions, 01_schemas, 02_sequences, 03_tables, 04_indexes, 05_procedures, 06_triggers, 07_reference_data, 08_script_log, 09_schema_check, 10_schema_version Every fresh install runs all of them, so every client sees the same database. Afterwards `SELECT * FROM meta.find_schema_problems()` must return no rows. Connection settings come from (highest precedence first): 1. command-line flags (--host --port --dbname --user --password) 2. db_config.json next to this script 3. built-in defaults (localhost:5432 / timeseries_db__dev / tsapi) Usage: python run_setup.py # use db_config.json / defaults python run_setup.py --host 192.168.1.7 --user claude --password claudepwd python run_setup.py --drop-existing # drop the existing installation first # (all series!); user schemas untouched A database that already holds the library's objects (meta, log, a non-empty series schema, the event trigger) is refused -- there are no migrations. User schemas and tables may exist and are never touched. """ import sys import json import argparse from pathlib import Path from typing import Dict SQL_DIR = Path(__file__).parent / "sql" CONFIG_FILE = Path(__file__).parent / "db_config.json" # Executed in order -- all of them, on every fresh install. 10_schema_version # runs last, so a database that records a version is complete. SQL_FILES = [ "00_extensions.sql", "01_schemas.sql", "02_sequences.sql", "03_tables.sql", "04_indexes.sql", "05_procedures.sql", "06_triggers.sql", "07_reference_data.sql", "08_script_log.sql", "09_schema_check.sql", "10_schema_version.sql", ] DEFAULTS = { "host": "localhost", "port": 5432, "database": "timeseries_db__dev", "user": "tsapi", "password": "tsapipwd", } def load_config(args) -> Dict: config = dict(DEFAULTS) if CONFIG_FILE.exists(): try: config.update(json.loads(CONFIG_FILE.read_text())) except json.JSONDecodeError as e: print(f"[FAIL] Invalid JSON in {CONFIG_FILE}: {e}") sys.exit(1) # CLI overrides for key, val in ( ("host", args.host), ("port", args.port), ("database", args.dbname), ("user", args.user), ("password", args.password), ): if val is not None: config[key] = val return config def execute_sql_file(conn, sql_file: Path): try: sql = sql_file.read_text(encoding="utf-8") with conn.cursor() as cur: cur.execute(sql) # committed by the caller, all files at once return True, f"[OK] Executed {sql_file.name}" except Exception as e: conn.rollback() return False, f"[FAIL] {sql_file.name}: {e}" def verify_setup(conn) -> None: print("\n" + "=" * 70) print("Verifying setup") print("=" * 70) with conn.cursor() as cur: cur.execute( "SELECT schema_name FROM information_schema.schemata " "WHERE schema_name IN ('meta','series','log') ORDER BY 1" ) print("[OK] Schemas:", ", ".join(r[0] for r in cur.fetchall())) cur.execute("SELECT count(*) FROM information_schema.tables WHERE table_schema='meta'") print("[OK] meta tables:", cur.fetchone()[0]) cur.execute("SELECT count(*) FROM information_schema.routines WHERE routine_schema IN ('meta','log')") print("[OK] procedures:", cur.fetchone()[0]) cur.execute("SELECT count(*) FROM pg_event_trigger WHERE evtname='prevent_series_drops'") print("[OK] event triggers:", cur.fetchone()[0]) cur.execute("SELECT count(*) FROM meta.timezones") tz = cur.fetchone()[0] cur.execute("SELECT count(*) FROM meta.currencies") cu = cur.fetchone()[0] print(f"[OK] reference data: {tz} timezones, {cu} currencies") cur.execute("SELECT version, description FROM meta.schema_version ORDER BY installed_at DESC LIMIT 1") row = cur.fetchone() if row: print(f"[OK] schema_version: {row[0]} ({row[1]})") # Pre-check, shared rule with the Python client's init_postgres_db(): # provisioning needs the library's own objects to be ABSENT. User schemas and # tables are irrelevant and never touched. Never drop-and-recreate: an # existing installation is refused, because there are no migrations and # re-provisioning deletes all series. PRECHECK_SQL = """ SELECT 'schema ' || nspname || ' exists' FROM pg_namespace WHERE nspname IN ('meta', 'log') UNION ALL SELECT 'schema series is not empty (' || count(*) || ' objects)' FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace WHERE n.nspname = 'series' AND c.relkind IN ('r', 'p', 'v', 'm', 'f', 'S') HAVING count(*) > 0 UNION ALL SELECT 'event trigger prevent_series_drops exists' FROM pg_event_trigger WHERE evtname = 'prevent_series_drops' """ def existing_installation(conn) -> list: """What would stop a fresh install; [] means go ahead.""" with conn.cursor() as cur: cur.execute(PRECHECK_SQL) found = [r[0] for r in cur.fetchall()] cur.execute("SELECT to_regclass('meta.schema_version') IS NOT NULL") if cur.fetchone()[0]: cur.execute("SELECT string_agg(version, ', ') FROM meta.schema_version") version = cur.fetchone()[0] found.append(f"installed schema version: {version or 'none recorded'}") return found def drop_existing(conn) -> None: """Drop the library's objects: every registered series schema, meta, log, the event trigger. Refuses if a series schema holds non-series objects -- those are user data, which this never deletes.""" with conn.cursor() as cur: schemas = ['series'] cur.execute("SELECT to_regclass('meta.series_schemas') IS NOT NULL") if cur.fetchone()[0]: cur.execute("SELECT name FROM meta.series_schemas") schemas = sorted({r[0] for r in cur.fetchall()} | {'series'}) cur.execute("SELECT to_regclass('meta.series') IS NOT NULL") has_registry = cur.fetchone()[0] stray = [] for schema in schemas: cur.execute( "SELECT c.relname FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace " "WHERE n.nspname = %s AND c.relkind IN ('r','p','v','m','f','S')", (schema,)) names = [r[0] for r in cur.fetchall()] if has_registry: cur.execute("SELECT table_name FROM meta.series WHERE db_schema = %s", (schema,)) names = sorted(set(names) - {r[0] for r in cur.fetchall()}) stray += [f"{schema}.{n}" for n in names] if stray: raise RuntimeError( "refusing to drop: these objects in series schemas are not series " f"(user data?): {', '.join(stray)}. Move or drop them first.") cur.execute("DROP EVENT TRIGGER IF EXISTS prevent_series_drops") for schema in schemas: cur.execute(f"DROP SCHEMA IF EXISTS {schema} CASCADE") print(f" dropped schema {schema}") cur.execute("DROP SCHEMA IF EXISTS meta CASCADE") cur.execute("DROP SCHEMA IF EXISTS log CASCADE") conn.commit() print("[OK] existing installation dropped\n") def main() -> int: p = argparse.ArgumentParser(description="Provision the TSAPI series DB schema") p.add_argument("--host") p.add_argument("--port", type=int) p.add_argument("--dbname") p.add_argument("--user") p.add_argument("--password") p.add_argument("--drop-existing", action="store_true", help="Drop the existing installation first: every series schema " "(ALL series data), meta, log. User schemas are untouched.") p.add_argument("--yes", action="store_true", help="Skip confirmation prompts") args = p.parse_args() try: import psycopg2 except ModuleNotFoundError: print("[FAIL] psycopg2 not installed. Run inside an environment that has " "it (e.g. the TSAPI_SERIESDB_PG_PY venv), or: pip install psycopg2-binary") return 1 cfg = load_config(args) print("=" * 70) print("TSAPI series DB schema setup") print("=" * 70) print(f"Target : {cfg['host']}:{cfg['port']}/{cfg['database']} (user {cfg['user']})") print(f"SQL dir: {SQL_DIR}") if args.drop_existing and not args.yes: what = "the existing installation, including ALL series data" if input(f"This will DROP {what}. Continue? (yes/no): ").lower() != "yes": print("Aborted.") return 1 try: conn = psycopg2.connect( host=cfg["host"], port=cfg["port"], dbname=cfg["database"], user=cfg["user"], password=cfg["password"], ) except Exception as e: print(f"[FAIL] connect: {e}") return 1 try: if args.drop_existing: try: drop_existing(conn) except RuntimeError as e: print(f"[FAIL] {e}") return 1 existing = existing_installation(conn) if existing: print("[FAIL] the database already holds a tsapi installation:") for line in existing: print(f" - {line}") print(" There are no migrations. Re-provision with --drop-existing " "(deletes all series), or use another database.") return 1 missing = [n for n in SQL_FILES if not (SQL_DIR / n).exists()] if missing: print(f"[FAIL] missing {', '.join(missing)}") return 1 # One transaction: a failure part-way leaves the database untouched. for name in SQL_FILES: ok, msg = execute_sql_file(conn, SQL_DIR / name) print(msg) if not ok: print(f"\n[FAIL] {name} failed; nothing was installed") return 1 conn.commit() verify_setup(conn) print("\n[SUCCESS] schema setup complete") return 0 finally: conn.close() if __name__ == "__main__": sys.exit(main())