Codex
82e2c91f42
Publish documentation / publish (push) Successful in 1m27s
Add PostgreSQL-backed memory, editable evidence with source review and activation, and human-approved archive repairs across the harness, API, and UI. Include migrations, deployment support, regression coverage, and validation documentation. Refresh permissions from validated session roles so existing administrator logins can access newly deployed archive management features.
75 lines
3.1 KiB
Python
75 lines
3.1 KiB
Python
"""Versioned Memory migration pack, run only by installation preparation."""
|
|
|
|
import hashlib
|
|
import os
|
|
from functools import lru_cache
|
|
from importlib.resources import files
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import URL, create_engine, text
|
|
from sqlalchemy.pool import NullPool
|
|
|
|
|
|
def installation_url(*, migrator: bool = False) -> str:
|
|
role = "MIGRATOR" if migrator else "RUNTIME"
|
|
direct = os.environ.get(f"THT_CATALOG_{role}_DATABASE_URL")
|
|
if not migrator:
|
|
direct = direct or os.environ.get("THT_CATALOG_DATABASE_URL")
|
|
if direct:
|
|
return direct.replace("postgresql://", "postgresql+psycopg2://", 1)
|
|
prefix = "THT_CATALOG_"
|
|
try:
|
|
password = Path(os.environ[prefix + role + "_PASSWORD_FILE"]).read_text().strip()
|
|
url = URL.create(
|
|
"postgresql+psycopg2", host=os.environ[prefix + "DB_HOST"],
|
|
port=int(os.environ.get(prefix + "DB_PORT", "5432")),
|
|
database=os.environ[prefix + "DB_NAME"],
|
|
username=os.environ[prefix + role + "_USER"], password=password,
|
|
)
|
|
return url.render_as_string(hide_password=False)
|
|
except (KeyError, OSError, ValueError):
|
|
raise ValueError("Memory PostgreSQL installation configuration is unavailable") from None
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def expected_migrations() -> dict[str, str]:
|
|
return {p.name: hashlib.sha256(p.read_text().encode()).hexdigest()
|
|
for p in files("tht").joinpath("migrations/memory").iterdir()
|
|
if p.name.endswith(".sql")}
|
|
|
|
|
|
def migrate(database_url: str) -> None:
|
|
engine = create_engine(database_url, poolclass=NullPool)
|
|
try:
|
|
with engine.begin() as connection:
|
|
connection.execute(text("SELECT pg_advisory_xact_lock(792114203)"))
|
|
connection.execute(text("CREATE SCHEMA IF NOT EXISTS thoth_memory"))
|
|
connection.execute(text("CREATE TABLE IF NOT EXISTS thoth_memory.migrations "
|
|
"(version text PRIMARY KEY, checksum text NOT NULL)"))
|
|
applied = dict(connection.execute(text(
|
|
"SELECT version, checksum FROM thoth_memory.migrations"
|
|
)).all())
|
|
pack = sorted(files("tht").joinpath("migrations/memory").iterdir(), key=lambda p: p.name)
|
|
known = {p.name for p in pack if p.name.endswith(".sql")}
|
|
if set(applied) - known:
|
|
raise ValueError("Memory schema is newer than this application")
|
|
for path in pack:
|
|
if path.name not in known:
|
|
continue
|
|
sql = path.read_text()
|
|
digest = hashlib.sha256(sql.encode()).hexdigest()
|
|
if path.name in applied:
|
|
if applied[path.name] != digest:
|
|
raise ValueError("Memory migration checksum mismatch")
|
|
continue
|
|
connection.execute(text(sql))
|
|
connection.execute(text("INSERT INTO thoth_memory.migrations VALUES (:v, :c)"),
|
|
{"v": path.name, "c": digest})
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
migrate(installation_url(migrator=True))
|
|
print("Memory migrations: ready")
|