peter
9032fd3a45
- Removed --clean-slate: it dropped every schema in the database, including user data. Nothing used it. - Pre-check (same rule as the Python client's init_postgres_db()): refuse if meta or log exists, series is not empty, or the prevent_series_drops trigger exists, naming the installed version. User schemas and tables are allowed and never touched; an empty series schema is reused. - --drop-existing drops every registered series schema plus meta and log, and refuses if a series schema holds non-series objects. - All files run in one transaction: a failure leaves nothing behind. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
276 lines
10 KiB
Python
276 lines
10 KiB
Python
"""
|
|
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())
|