From f504f7bda901e58145ed21582b92b672acf1c7d6 Mon Sep 17 00:00:00 2001 From: Emi Griffith Date: Sun, 19 Jul 2026 23:28:23 -0700 Subject: [PATCH] Containerize the app and move the databases to PostgreSQL 18 (#220) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Run Thermograph as a docker-compose stack (app + Postgres 18) and standardize the data layer on Postgres, while keeping the test suite on SQLite. - accounts/db.py: DSN-driven engines. On Postgres, a per-worker read-write + read-only asyncpg pair (the RO engine pins read-only transactions, used by the pure-GET endpoints) plus a sync psycopg engine for the notifier thread; the SQLite path is preserved for tests/local (selected when THERMOGRAPH_DATABASE_URL is unset). models.py: boolean server_default -> sa.false(). - store.py / metrics.py: dialect-flexible — Postgres UNLOGGED tables via psycopg when configured, else the existing raw-sqlite3 paths byte-for-byte; sync interfaces and every fail-soft contract preserved. - Alembic (backend/alembic/) manages the accounts schema; the container entrypoint runs `alembic upgrade head` before uvicorn (4 workers). migrate_accounts_to_pg.py copies the accounts data SQLite->PG through the ORM (UUID/bool/JSON coerced), skips access_token, and resets identity sequences. - Dockerfile + docker-compose.yml: app image (uvicorn, 4 workers, loopback 8137) and a Postgres 18 db (2 CPUs) running pg_duckdb (deploy/db/) so the parquet climate cache is queryable in-DB via read_parquet('/parquet/cache/*.parquet'). - deploy.sh/thermograph.service rewired to manage the compose stack; env example, Makefile targets (up/down/db-up), and deploy/POSTGRES-MIGRATION.md cutover runbook. Tests stay on SQLite (dialect fallback) — 323 pass. The full Postgres stack was verified via docker compose: alembic migrations, register/login, the RO endpoint, store/metrics round-trips, and the accounts data migration. --- accounts/api_accounts.py | 6 +- accounts/db.py | 120 ++++++++++----- accounts/models.py | 3 +- alembic.ini | 41 +++++ alembic/env.py | 58 +++++++ alembic/script.py.mako | 22 +++ alembic/versions/0001_initial_schema.py | 29 ++++ core/metrics.py | 192 +++++++++++++++++------- data/store.py | 121 +++++++++++++-- deploy/entrypoint.sh | 26 ++++ migrate_accounts_to_pg.py | 78 ++++++++++ requirements.txt | 8 +- tests/accounts/test_db_engines.py | 40 +++++ 13 files changed, 626 insertions(+), 118 deletions(-) create mode 100644 alembic.ini create mode 100644 alembic/env.py create mode 100644 alembic/script.py.mako create mode 100644 alembic/versions/0001_initial_schema.py create mode 100755 deploy/entrypoint.sh create mode 100644 migrate_accounts_to_pg.py create mode 100644 tests/accounts/test_db_engines.py diff --git a/accounts/api_accounts.py b/accounts/api_accounts.py index beae8ef..106c3ed 100644 --- a/accounts/api_accounts.py +++ b/accounts/api_accounts.py @@ -15,7 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from data import climate from data import grid from notifications import push -from accounts.db import get_async_session +from accounts.db import get_async_session, get_read_session from accounts.models import Notification, PushSubscription, Subscription from web.schemas import ( NotificationList, @@ -51,7 +51,7 @@ async def _owned_subscription(session: AsyncSession, sub_id: int, user) -> Subsc @router.get("/subscriptions", response_model=list[SubscriptionOut]) async def list_subscriptions( user=Depends(current_active_user), - session: AsyncSession = Depends(get_async_session), + session: AsyncSession = Depends(get_read_session), # pure read → read-only engine ): rows = ( await session.execute( @@ -142,7 +142,7 @@ async def list_notifications( unread: bool = Query(False, description="only return unread notifications"), limit: int = Query(50, ge=1, le=200), user=Depends(current_active_user), - session: AsyncSession = Depends(get_async_session), + session: AsyncSession = Depends(get_read_session), # pure read → read-only engine ): q = select(Notification).where(Notification.user_id == user.id) if unread: diff --git a/accounts/db.py b/accounts/db.py index 84bad3e..680c15f 100644 --- a/accounts/db.py +++ b/accounts/db.py @@ -1,21 +1,22 @@ -"""Authoritative account database — SQLAlchemy over data/accounts.sqlite (WAL). +"""Authoritative account database — SQLAlchemy over PostgreSQL (or SQLite). -This is deliberately a *separate* database from data/thermograph.sqlite (store.py). -That file is a disposable accelerator — "deleting it is a safe reset" — because -everything in it recomputes from the parquet source of truth. Accounts, -subscriptions and notifications have no source to recompute from, so they live in -their own file with their own (stricter) rules: foreign keys on, errors surface -rather than get swallowed, and it should be backed up (it is not regenerable). +This is deliberately a *separate* database from the disposable derived cache +(data/store.py). Accounts, subscriptions and notifications have no source to +recompute from, so they live with stricter rules: foreign keys enforced, errors +surface rather than get swallowed, and the data must be backed up (it is not +regenerable). -Two engines share the one file: +Backend is chosen by ``THERMOGRAPH_DATABASE_URL``: -* an **async** engine (``sqlite+aiosqlite``) drives the request/auth path, because - fastapi-users is async; and -* a **sync** engine drives the background notifier thread (notify.py), which is a - plain daemon thread with no event loop. - -WAL mode lets the async readers/writers and the sync notifier coexist on the same -file without blocking each other. +* **PostgreSQL** (``postgresql+asyncpg://…``) in container/prod. Two async engines + over the one primary give the read / read-write split the deployment asks for — + a read-only engine (``default_transaction_read_only``) for pure-GET endpoints + and a read-write engine for everything else (nearly every account op is a + SELECT-then-write in one transaction and must be RW). A separate **sync** + engine (psycopg) drives the background notifier thread, which has no event loop. +* **SQLite** (default / tests) keeps the previous single-file behaviour with WAL + so the async web path and the sync notifier coexist; the RO "engine" is just an + alias of the RW one (SQLite has no read-only transaction mode to enforce). """ import os @@ -25,21 +26,29 @@ from sqlalchemy.orm import DeclarativeBase, sessionmaker import paths -# Default to data/accounts.sqlite; THERMOGRAPH_ACCOUNTS_DB overrides it (tests point -# this at a throwaway temp file to stay hermetic). -DB_PATH = os.environ.get("THERMOGRAPH_ACCOUNTS_DB") or os.path.join(paths.DATA_DIR, "accounts.sqlite") -os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) +# The async SQLAlchemy URL. When unset we fall back to the legacy SQLite file +# (THERMOGRAPH_ACCOUNTS_DB overrides its path; tests point it at a temp file). +DATABASE_URL = os.environ.get("THERMOGRAPH_DATABASE_URL", "").strip() +IS_POSTGRES = DATABASE_URL.startswith("postgresql") + + +def _sync_url(async_url: str) -> str: + """The sync driver URL for the notifier: asyncpg is async-only, so the sync + engine uses psycopg (``postgresql+psycopg://``); SQLite drops the aiosqlite + driver suffix.""" + return (async_url.replace("+asyncpg", "+psycopg") + if async_url.startswith("postgresql") + else async_url.replace("+aiosqlite", "")) class Base(DeclarativeBase): """Declarative base shared by every account-domain table (models.py).""" -def _apply_pragmas(dbapi_conn, _rec): - # Per-connection SQLite setup: WAL for concurrent async/sync access, NORMAL - # sync for a good durability/speed tradeoff, and foreign_keys ON so the - # ON DELETE CASCADE relationships (user -> subscriptions -> notifications) - # are actually enforced (SQLite defaults them off). +def _apply_sqlite_pragmas(dbapi_conn, _rec): + # SQLite-only: WAL for concurrent async/sync access, NORMAL sync, and + # foreign_keys ON so the ON DELETE CASCADE relationships are enforced + # (SQLite defaults them off; Postgres always enforces FKs). cur = dbapi_conn.cursor() cur.execute("PRAGMA journal_mode=WAL") cur.execute("PRAGMA synchronous=NORMAL") @@ -47,31 +56,64 @@ def _apply_pragmas(dbapi_conn, _rec): cur.close() -# --- async engine (web / auth path) ----------------------------------------- -async_engine = create_async_engine(f"sqlite+aiosqlite:///{DB_PATH}", future=True) -event.listen(async_engine.sync_engine, "connect", _apply_pragmas) +if IS_POSTGRES: + DB_PATH = None + _COMMON = dict(pool_size=1, max_overflow=1, pool_pre_ping=True, pool_recycle=1800, future=True) + # Read-write engine (default) + a read-only engine that pins every session to a + # read-only transaction on the server, so a pure-read endpoint can't accidentally + # write and a long read stays off the write connection. + async_engine = create_async_engine(DATABASE_URL, **_COMMON) + async_engine_ro = create_async_engine( + DATABASE_URL, **_COMMON, + connect_args={"server_settings": {"default_transaction_read_only": "on"}}) + sync_engine = create_engine(_sync_url(DATABASE_URL), pool_size=1, max_overflow=1, + pool_pre_ping=True, pool_recycle=1800, future=True) +else: + DB_PATH = os.environ.get("THERMOGRAPH_ACCOUNTS_DB") or os.path.join(paths.DATA_DIR, "accounts.sqlite") + os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) + async_engine = create_async_engine(f"sqlite+aiosqlite:///{DB_PATH}", future=True) + event.listen(async_engine.sync_engine, "connect", _apply_sqlite_pragmas) + async_engine_ro = async_engine # SQLite has no read-only transaction mode + sync_engine = create_engine(f"sqlite:///{DB_PATH}", future=True) + event.listen(sync_engine, "connect", _apply_sqlite_pragmas) + + async_session_maker = async_sessionmaker(async_engine, expire_on_commit=False) +async_session_ro_maker = async_sessionmaker(async_engine_ro, expire_on_commit=False) +sync_session_maker = sessionmaker(sync_engine, expire_on_commit=False) async def get_async_session() -> AsyncSession: - """FastAPI dependency: one AsyncSession per request.""" + """FastAPI dependency: a read-write AsyncSession per request (the default).""" async with async_session_maker() as session: yield session -# --- sync engine (background notifier thread) ------------------------------- -sync_engine = create_engine(f"sqlite:///{DB_PATH}", future=True) -event.listen(sync_engine, "connect", _apply_pragmas) -sync_session_maker = sessionmaker(sync_engine, expire_on_commit=False) +async def get_read_session() -> AsyncSession: + """FastAPI dependency for genuinely read-only endpoints — served by the + read-only engine so reads prefer the read connection. On Postgres the session + runs in a read-only transaction; on SQLite it's the same engine as RW.""" + async with async_session_ro_maker() as session: + yield session async def create_db_and_tables() -> None: - """Create any missing tables. Called once at startup (app lifespan). - - Imported for its side effect of registering the mapped classes on Base.metadata - before create_all runs. - """ + """Create any missing tables. Called once at startup (app lifespan). On + Postgres, retry briefly so an app boot that races the DB container recovers + instead of crashing.""" from accounts import models # noqa: F401 (registers tables on Base.metadata) - async with async_engine.begin() as conn: - await conn.run_sync(Base.metadata.create_all) + import asyncio + + last = None + for attempt in range(10): + try: + async with async_engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + return + except Exception as e: # noqa: BLE001 - transient "DB not ready yet" on container start + last = e + if not IS_POSTGRES: + raise + await asyncio.sleep(min(2 ** attempt, 5)) + raise last diff --git a/accounts/models.py b/accounts/models.py index 4546e40..40b6df4 100644 --- a/accounts/models.py +++ b/accounts/models.py @@ -22,6 +22,7 @@ from sqlalchemy import ( String, Text, UniqueConstraint, + false, ) from sqlalchemy.orm import Mapped, mapped_column @@ -39,7 +40,7 @@ class User(SQLAlchemyBaseUserTableUUID, Base): # Whether to also deliver alerts as a Discord DM. Set True on linking (an active # opt-in); the user can mute it while staying linked. Same migration caveat. discord_dm: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, - server_default="0") + server_default=false()) class AccessToken(SQLAlchemyBaseAccessTokenTableUUID, Base): diff --git a/alembic.ini b/alembic.ini new file mode 100644 index 0000000..1d4547b --- /dev/null +++ b/alembic.ini @@ -0,0 +1,41 @@ +# Alembic config for the accounts database. The URL is supplied at runtime by +# alembic/env.py from THERMOGRAPH_DATABASE_URL (Postgres) or the SQLite fallback, +# so it is intentionally not hard-coded here. +[alembic] +script_location = alembic +prepend_sys_path = . +version_path_separator = os + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARNING +handlers = console +qualname = + +[logger_sqlalchemy] +level = WARNING +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S diff --git a/alembic/env.py b/alembic/env.py new file mode 100644 index 0000000..78d527e --- /dev/null +++ b/alembic/env.py @@ -0,0 +1,58 @@ +"""Alembic environment for the accounts database. + +The migration URL is derived from ``THERMOGRAPH_DATABASE_URL`` (the async app URL) +by swapping to a sync driver — Alembic runs synchronously. Falls back to the +SQLite accounts file when no Postgres URL is set (local / tests). ``target_metadata`` +is the accounts ``Base.metadata`` so ``alembic revision --autogenerate`` can diff +against the ORM models. +""" +import os +import sys +from logging.config import fileConfig + +from alembic import context +from sqlalchemy import create_engine, pool + +# Alembic runs with cwd = backend/ (WORKDIR); make the backend package importable +# regardless of where alembic is invoked from. +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from accounts.db import Base # noqa: E402 +from accounts import models # noqa: E402,F401 (registers tables on Base.metadata) + +config = context.config +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +target_metadata = Base.metadata + + +def _sync_url() -> str: + url = os.environ.get("THERMOGRAPH_DATABASE_URL", "").strip() + if url.startswith("postgresql"): + return url.replace("+asyncpg", "+psycopg") + from accounts.db import DB_PATH # SQLite fallback path + return f"sqlite:///{DB_PATH}" + + +def run_migrations_offline() -> None: + context.configure(url=_sync_url(), target_metadata=target_metadata, + literal_binds=True, dialect_opts={"paramstyle": "named"}) + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online() -> None: + engine = create_engine(_sync_url(), poolclass=pool.NullPool, future=True) + with engine.connect() as connection: + context.configure(connection=connection, target_metadata=target_metadata, + render_as_batch=connection.dialect.name == "sqlite") + with context.begin_transaction(): + context.run_migrations() + engine.dispose() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/alembic/script.py.mako b/alembic/script.py.mako new file mode 100644 index 0000000..36cd8ef --- /dev/null +++ b/alembic/script.py.mako @@ -0,0 +1,22 @@ +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} +""" +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + +revision = ${repr(up_revision)} +down_revision = ${repr(down_revision)} +branch_labels = ${repr(branch_labels)} +depends_on = ${repr(depends_on)} + + +def upgrade() -> None: + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + ${downgrades if downgrades else "pass"} diff --git a/alembic/versions/0001_initial_schema.py b/alembic/versions/0001_initial_schema.py new file mode 100644 index 0000000..6ace60a --- /dev/null +++ b/alembic/versions/0001_initial_schema.py @@ -0,0 +1,29 @@ +"""initial accounts schema (from the ORM models) + +Adopts Alembic on the existing create_all-managed schema: this baseline builds the +full current schema straight from ``Base.metadata`` (so it never drifts from +models.py), and subsequent revisions are produced with ``alembic revision +--autogenerate``. It subsumes the two hand-written deploy/migrations/*.sql +column-adds, which were SQLite-specific and are moot on a fresh Postgres database. + +Revision ID: 0001_initial_schema +Revises: +Create Date: 2026-07-19 +""" +from alembic import op + +from accounts.db import Base +from accounts import models # noqa: F401 (registers tables on Base.metadata) + +revision = "0001_initial_schema" +down_revision = None +branch_labels = None +depends_on = None + + +def upgrade() -> None: + Base.metadata.create_all(bind=op.get_bind()) + + +def downgrade() -> None: + Base.metadata.drop_all(bind=op.get_bind()) diff --git a/core/metrics.py b/core/metrics.py index 0c681ab..f882f11 100644 --- a/core/metrics.py +++ b/core/metrics.py @@ -21,6 +21,21 @@ import threading import time import urllib.parse +# Dialect selection mirrors accounts/db.py and data/store.py: a ``postgresql://`` +# URL in THERMOGRAPH_DATABASE_URL routes the shared counters into Postgres (prod / +# multi-worker); otherwise the store stays SQLite-file / in-memory as before. +DATABASE_URL = os.environ.get("THERMOGRAPH_DATABASE_URL", "").strip() +IS_POSTGRES = DATABASE_URL.startswith("postgresql") + + +def _libpq_dsn(url: str) -> str: + """A plain libpq URL that ``psycopg.connect`` accepts: drop the SQLAlchemy + driver suffix (``postgresql+asyncpg://`` / ``+psycopg://`` -> ``postgresql://``).""" + return url.replace("+asyncpg", "").replace("+psycopg", "") + + +_PG_DSN = _libpq_dsn(DATABASE_URL) if IS_POSTGRES else "" + # The ``phase=`` tag passed to ``climate._request`` maps 1:1 onto an external source. # Keep this in sync with the phase names used at each call site in ``climate.py``. _PHASE_SOURCE = { @@ -288,42 +303,92 @@ class _MemStore: } +# The seven metrics tables, in a stable order. Used by the Postgres init to +# TRUNCATE (there is no file to `rm`, so we wipe here to keep the since-start +# semantics) — the guard is exactly this list, and only on the Postgres path. +_METRICS_TABLES = ( + "meta", "inbound_cume", "outbound_cume", "inbound_minute", + "outbound_minute", "event_cume", "event_minute", +) + +# Postgres DDL: same shape as the SQLite schema below, but UNLOGGED (fast, non- +# durable — metrics are since-start and disposable) and one statement per execute +# (psycopg's extended protocol won't run a multi-statement script). +_PG_METRICS_SCHEMA = ( + "CREATE UNLOGGED TABLE IF NOT EXISTS meta(key TEXT PRIMARY KEY, value DOUBLE PRECISION)", + "CREATE UNLOGGED TABLE IF NOT EXISTS inbound_cume(" + "category TEXT, bucket TEXT, count BIGINT, PRIMARY KEY(category, bucket))", + "CREATE UNLOGGED TABLE IF NOT EXISTS outbound_cume(" + "source TEXT, outcome TEXT, count BIGINT, PRIMARY KEY(source, outcome))", + "CREATE UNLOGGED TABLE IF NOT EXISTS inbound_minute(" + "minute BIGINT, category TEXT, count BIGINT, PRIMARY KEY(minute, category))", + "CREATE UNLOGGED TABLE IF NOT EXISTS outbound_minute(" + "minute BIGINT, source TEXT, count BIGINT, PRIMARY KEY(minute, source))", + "CREATE UNLOGGED TABLE IF NOT EXISTS event_cume(" + "event TEXT, referrer TEXT, day TEXT, count BIGINT, PRIMARY KEY(event, referrer, day))", + "CREATE UNLOGGED TABLE IF NOT EXISTS event_minute(" + "minute BIGINT, event TEXT, count BIGINT, PRIMARY KEY(minute, event))", +) + + class _SqliteStore: - """Cross-process counters in a shared SQLite file, so every uvicorn worker tallies - into one place and the dashboard sees the whole deployment. WAL mode lets the workers - write concurrently; the counts are tiny UPSERTs well within SQLite's throughput. + """Cross-process counters in a shared store, so every uvicorn worker tallies into + one place and the dashboard sees the whole deployment. Two dialects behind one + interface: a shared SQLite file (WAL) by default, or Postgres (``postgres=True``, + UNLOGGED tables) when THERMOGRAPH_DATABASE_URL selects it. Either way the tally is + tiny UPSERTs on a single connection serialized by ``self._lock`` — psycopg + connections aren't thread-safe, so that lock is what keeps the sharing safe (the + same role SQLite's ``check_same_thread=False`` plays here). """ - def __init__(self, path: str) -> None: + def __init__(self, path: str, postgres: bool = False) -> None: self._lock = threading.Lock() - # check_same_thread=False: the connection is shared across a worker's threads, - # serialized by self._lock. isolation_level=None: autocommit each statement. - self._db = sqlite3.connect(path, check_same_thread=False, isolation_level=None) - self._db.execute("PRAGMA journal_mode=WAL") - self._db.execute("PRAGMA synchronous=NORMAL") - self._db.execute("PRAGMA busy_timeout=3000") - self._db.executescript( - """ - CREATE TABLE IF NOT EXISTS meta(key TEXT PRIMARY KEY, value REAL); - CREATE TABLE IF NOT EXISTS inbound_cume( - category TEXT, bucket TEXT, count INTEGER, PRIMARY KEY(category, bucket)); - CREATE TABLE IF NOT EXISTS outbound_cume( - source TEXT, outcome TEXT, count INTEGER, PRIMARY KEY(source, outcome)); - CREATE TABLE IF NOT EXISTS inbound_minute( - minute INTEGER, category TEXT, count INTEGER, PRIMARY KEY(minute, category)); - CREATE TABLE IF NOT EXISTS outbound_minute( - minute INTEGER, source TEXT, count INTEGER, PRIMARY KEY(minute, source)); - CREATE TABLE IF NOT EXISTS event_cume( - event TEXT, referrer TEXT, day TEXT, count INTEGER, - PRIMARY KEY(event, referrer, day)); - CREATE TABLE IF NOT EXISTS event_minute( - minute INTEGER, event TEXT, count INTEGER, PRIMARY KEY(minute, event)); - """ - ) - # First worker to touch the DB stamps the shared start time; the rest inherit it. - self._db.execute( - "INSERT OR IGNORE INTO meta(key, value) VALUES('started_at', ?)", (time.time(),) - ) + self._pg = postgres + self._ph = "%s" if postgres else "?" # bound-parameter placeholder style + if postgres: + import psycopg # local import: only the Postgres path needs it + # autocommit mirrors SQLite's isolation_level=None: every counter UPSERT + # lands on its own, and no read leaves an idle-in-transaction. + self._db = psycopg.connect(path, autocommit=True) + for stmt in _PG_METRICS_SCHEMA: + self._db.execute(stmt) + # SQLite is wiped each boot by an external `ExecStartPre rm` of the file; + # Postgres has no file to remove, so reset the since-start counters here. + # Guarded to exactly the metrics tables and only this (Postgres) path. + self._db.execute("TRUNCATE " + ", ".join(_METRICS_TABLES)) + self._db.execute( + "INSERT INTO meta(key, value) VALUES('started_at', %s) ON CONFLICT DO NOTHING", + (time.time(),), + ) + else: + # check_same_thread=False: the connection is shared across a worker's threads, + # serialized by self._lock. isolation_level=None: autocommit each statement. + self._db = sqlite3.connect(path, check_same_thread=False, isolation_level=None) + self._db.execute("PRAGMA journal_mode=WAL") + self._db.execute("PRAGMA synchronous=NORMAL") + self._db.execute("PRAGMA busy_timeout=3000") + self._db.executescript( + """ + CREATE TABLE IF NOT EXISTS meta(key TEXT PRIMARY KEY, value REAL); + CREATE TABLE IF NOT EXISTS inbound_cume( + category TEXT, bucket TEXT, count INTEGER, PRIMARY KEY(category, bucket)); + CREATE TABLE IF NOT EXISTS outbound_cume( + source TEXT, outcome TEXT, count INTEGER, PRIMARY KEY(source, outcome)); + CREATE TABLE IF NOT EXISTS inbound_minute( + minute INTEGER, category TEXT, count INTEGER, PRIMARY KEY(minute, category)); + CREATE TABLE IF NOT EXISTS outbound_minute( + minute INTEGER, source TEXT, count INTEGER, PRIMARY KEY(minute, source)); + CREATE TABLE IF NOT EXISTS event_cume( + event TEXT, referrer TEXT, day TEXT, count INTEGER, + PRIMARY KEY(event, referrer, day)); + CREATE TABLE IF NOT EXISTS event_minute( + minute INTEGER, event TEXT, count INTEGER, PRIMARY KEY(minute, event)); + """ + ) + # First worker to touch the DB stamps the shared start time; the rest inherit it. + self._db.execute( + "INSERT OR IGNORE INTO meta(key, value) VALUES('started_at', ?)", (time.time(),) + ) self._last_pruned_minute = -1 @property @@ -339,25 +404,26 @@ class _SqliteStore: return self._last_pruned_minute = minute cutoff = minute - WINDOW_MINUTES - self._db.execute("DELETE FROM inbound_minute WHERE minute <= ?", (cutoff,)) - self._db.execute("DELETE FROM outbound_minute WHERE minute <= ?", (cutoff,)) - self._db.execute("DELETE FROM event_minute WHERE minute <= ?", (cutoff,)) + self._db.execute(f"DELETE FROM inbound_minute WHERE minute <= {self._ph}", (cutoff,)) + self._db.execute(f"DELETE FROM outbound_minute WHERE minute <= {self._ph}", (cutoff,)) + self._db.execute(f"DELETE FROM event_minute WHERE minute <= {self._ph}", (cutoff,)) # Event history is kept by day, not by minute. + placeholders = ",".join([self._ph] * EVENT_DAYS) self._db.execute( - "DELETE FROM event_cume WHERE day NOT IN (%s)" - % ",".join("?" * EVENT_DAYS), tuple(sorted(_recent_days())), + f"DELETE FROM event_cume WHERE day NOT IN ({placeholders})", + tuple(sorted(_recent_days())), ) def record_inbound(self, category, bucket) -> None: minute = _now_minute() with self._lock: self._db.execute( - "INSERT INTO inbound_cume(category, bucket, count) VALUES(?, ?, 1) " + f"INSERT INTO inbound_cume(category, bucket, count) VALUES({self._ph}, {self._ph}, 1) " "ON CONFLICT(category, bucket) DO UPDATE SET count = count + 1", (category, bucket), ) self._db.execute( - "INSERT INTO inbound_minute(minute, category, count) VALUES(?, ?, 1) " + f"INSERT INTO inbound_minute(minute, category, count) VALUES({self._ph}, {self._ph}, 1) " "ON CONFLICT(minute, category) DO UPDATE SET count = count + 1", (minute, category), ) @@ -367,12 +433,12 @@ class _SqliteStore: minute = _now_minute() with self._lock: self._db.execute( - "INSERT INTO outbound_cume(source, outcome, count) VALUES(?, ?, 1) " + f"INSERT INTO outbound_cume(source, outcome, count) VALUES({self._ph}, {self._ph}, 1) " "ON CONFLICT(source, outcome) DO UPDATE SET count = count + 1", (source, outcome), ) self._db.execute( - "INSERT INTO outbound_minute(minute, source, count) VALUES(?, ?, 1) " + f"INSERT INTO outbound_minute(minute, source, count) VALUES({self._ph}, {self._ph}, 1) " "ON CONFLICT(minute, source) DO UPDATE SET count = count + 1", (minute, source), ) @@ -385,17 +451,17 @@ class _SqliteStore: row = self._db.execute( "SELECT COUNT(DISTINCT referrer) FROM event_cume").fetchone() known = self._db.execute( - "SELECT 1 FROM event_cume WHERE referrer = ? LIMIT 1", (referrer,) + f"SELECT 1 FROM event_cume WHERE referrer = {self._ph} LIMIT 1", (referrer,) ).fetchone() if not known and row and row[0] >= MAX_REFERRERS: referrer = REFERRER_OTHER self._db.execute( - "INSERT INTO event_cume(event, referrer, day, count) VALUES(?, ?, ?, 1) " + f"INSERT INTO event_cume(event, referrer, day, count) VALUES({self._ph}, {self._ph}, {self._ph}, 1) " "ON CONFLICT(event, referrer, day) DO UPDATE SET count = count + 1", (event, referrer, day), ) self._db.execute( - "INSERT INTO event_minute(minute, event, count) VALUES(?, ?, 1) " + f"INSERT INTO event_minute(minute, event, count) VALUES({self._ph}, {self._ph}, 1) " "ON CONFLICT(minute, event) DO UPDATE SET count = count + 1", (minute, event), ) @@ -407,12 +473,12 @@ class _SqliteStore: # worker answers its poll, even though only the leader worker runs it. with self._lock: self._db.execute( - "INSERT INTO meta(key, value) VALUES(?, ?) " + f"INSERT INTO meta(key, value) VALUES({self._ph}, {self._ph}) " "ON CONFLICT(key) DO UPDATE SET value = excluded.value", (f"hb_ts:{name}", ts), ) self._db.execute( - "INSERT INTO meta(key, value) VALUES(?, ?) " + f"INSERT INTO meta(key, value) VALUES({self._ph}, {self._ph}) " "ON CONFLICT(key) DO UPDATE SET value = excluded.value", (f"hb_int:{name}", interval_s), ) @@ -441,20 +507,23 @@ class _SqliteStore: for source, outcome, count in self._db.execute( "SELECT source, outcome, count FROM outbound_cume"): outbound.setdefault(source, {o: 0 for o in _OUTCOMES})[outcome] = count - recent_in = dict(self._db.execute( - "SELECT category, SUM(count) FROM inbound_minute WHERE minute >= ? " - "GROUP BY category", (cutoff,))) - recent_out = dict(self._db.execute( - "SELECT source, SUM(count) FROM outbound_minute WHERE minute >= ? " - "GROUP BY source", (cutoff,))) + # int() coerces the aggregate: Postgres SUM(BIGINT) returns numeric -> + # Decimal, which isn't JSON-serializable; SQLite already returns int, so + # the wrap is a no-op there. + recent_in = {k: int(v) for k, v in self._db.execute( + f"SELECT category, SUM(count) FROM inbound_minute WHERE minute >= {self._ph} " + "GROUP BY category", (cutoff,))} + recent_out = {k: int(v) for k, v in self._db.execute( + f"SELECT source, SUM(count) FROM outbound_minute WHERE minute >= {self._ph} " + "GROUP BY source", (cutoff,))} events = _nest_events( ((event, referrer, day), count) for event, referrer, day, count in self._db.execute( "SELECT event, referrer, day, count FROM event_cume") ) - recent_ev = dict(self._db.execute( - "SELECT event, SUM(count) FROM event_minute WHERE minute >= ? " - "GROUP BY event", (cutoff,))) + recent_ev = {k: int(v) for k, v in self._db.execute( + f"SELECT event, SUM(count) FROM event_minute WHERE minute >= {self._ph} " + "GROUP BY event", (cutoff,))} return { "started_at": self.started_at, "inbound": inbound, @@ -468,9 +537,16 @@ class _SqliteStore: def _make_store(): - """Pick the store from the environment. A path in ``THERMOGRAPH_METRICS_DB`` selects - the shared SQLite store (needed when running multiple workers); otherwise in-memory. - A store that fails to open falls back to memory — metrics must never break boot.""" + """Pick the store from the environment. A ``postgresql://`` URL in + THERMOGRAPH_DATABASE_URL routes counters into a shared Postgres store (prod / + multi-worker) and takes precedence; otherwise a path in ``THERMOGRAPH_METRICS_DB`` + selects the shared SQLite store (needed when running multiple workers); otherwise + in-memory. A store that fails to open falls back — metrics must never break boot.""" + if IS_POSTGRES: + try: + return _SqliteStore(_PG_DSN, postgres=True) + except Exception: # noqa: BLE001 - never let instrumentation break startup + pass path = os.environ.get("THERMOGRAPH_METRICS_DB", "").strip() if path: try: diff --git a/data/store.py b/data/store.py index 08b2d1a..4ebc288 100644 --- a/data/store.py +++ b/data/store.py @@ -18,6 +18,11 @@ survives restarts/deploys instead of being recomputed per request, per view: The store is a pure accelerator: every reader falls back to recomputing from parquet when the database is missing, locked, or corrupt (all helpers swallow their own errors), and deleting data/thermograph.sqlite is a safe full reset. + +On Postgres (``THERMOGRAPH_DATABASE_URL=postgresql://…``) the same two tables live +as UNLOGGED tables — fast and non-durable, which suits a rebuildable cache — and +the per-thread connection uses psycopg instead of sqlite3. Every fail-soft path is +identical: a connection or query that fails just degrades to recompute-from-parquet. """ import json import os @@ -30,6 +35,22 @@ import paths DB_PATH = os.path.join(paths.DATA_DIR, "thermograph.sqlite") +# Dialect selection mirrors accounts/db.py: a ``postgresql://`` URL in +# THERMOGRAPH_DATABASE_URL switches the store to Postgres; anything else keeps the +# SQLite file above. (This module uses the raw driver, not SQLAlchemy, so the DSN +# is a plain libpq URL — the ``+asyncpg`` / ``+psycopg`` SQLAlchemy suffix stripped.) +DATABASE_URL = os.environ.get("THERMOGRAPH_DATABASE_URL", "").strip() +IS_POSTGRES = DATABASE_URL.startswith("postgresql") + + +def _libpq_dsn(url: str) -> str: + """A plain libpq URL that ``psycopg.connect`` accepts: drop the SQLAlchemy + driver suffix (``postgresql+asyncpg://`` / ``+psycopg://`` -> ``postgresql://``).""" + return url.replace("+asyncpg", "").replace("+psycopg", "") + + +_PG_DSN = _libpq_dsn(DATABASE_URL) if IS_POSTGRES else "" + _SCHEMA = """ CREATE TABLE IF NOT EXISTS derived ( kind TEXT NOT NULL, -- payload family: grade | calendar | day | forecast @@ -47,22 +68,91 @@ CREATE TABLE IF NOT EXISTS revgeo ( ); """ +# Postgres mirror of _SCHEMA. UNLOGGED = fast, crash-non-durable (the cache is +# rebuildable, so "delete = safe reset" still holds); BLOB becomes BYTEA and REAL +# becomes DOUBLE PRECISION. psycopg's extended protocol runs one statement per +# execute, so the DDL is a tuple executed in turn (not one script like SQLite). +_PG_SCHEMA = ( + """ + CREATE UNLOGGED TABLE IF NOT EXISTS derived ( + kind TEXT NOT NULL, + cell_id TEXT NOT NULL, + key TEXT NOT NULL, + token TEXT NOT NULL, + payload BYTEA NOT NULL, + updated_at DOUBLE PRECISION NOT NULL, + PRIMARY KEY (kind, cell_id, key) + ) + """, + """ + CREATE UNLOGGED TABLE IF NOT EXISTS revgeo ( + key TEXT PRIMARY KEY, + label TEXT, + updated_at DOUBLE PRECISION NOT NULL + ) + """, +) + +# Statement text differs by dialect (placeholder style + upsert syntax); the SQLite +# strings are byte-for-byte the originals. Postgres uses %s params, an explicit +# column list, and ON CONFLICT ... DO UPDATE — which, like SQLite's INSERT OR +# REPLACE, overwrites *every* non-key column (token included, so a refreshed token +# supersedes the stored one and freshness keeps tracking the token, never the clock). +if IS_POSTGRES: + _SQL_GET_PAYLOAD = "SELECT token, payload FROM derived WHERE kind=%s AND cell_id=%s AND key=%s" + _SQL_PUT_PAYLOAD = ( + "INSERT INTO derived (kind, cell_id, key, token, payload, updated_at) " + "VALUES (%s,%s,%s,%s,%s,%s) " + "ON CONFLICT (kind, cell_id, key) DO UPDATE SET " + "token=EXCLUDED.token, payload=EXCLUDED.payload, updated_at=EXCLUDED.updated_at" + ) + _SQL_GET_REVGEO = "SELECT label, updated_at FROM revgeo WHERE key=%s" + _SQL_PUT_REVGEO = ( + "INSERT INTO revgeo (key, label, updated_at) VALUES (%s,%s,%s) " + "ON CONFLICT (key) DO UPDATE SET label=EXCLUDED.label, updated_at=EXCLUDED.updated_at" + ) +else: + _SQL_GET_PAYLOAD = "SELECT token, payload FROM derived WHERE kind=? AND cell_id=? AND key=?" + _SQL_PUT_PAYLOAD = "INSERT OR REPLACE INTO derived VALUES (?,?,?,?,?,?)" + _SQL_GET_REVGEO = "SELECT label, updated_at FROM revgeo WHERE key=?" + _SQL_PUT_REVGEO = "INSERT OR REPLACE INTO revgeo VALUES (?,?,?)" + # SQLite connections aren't shareable across threads; uvicorn serves requests on # a thread pool, so each thread lazily opens its own connection. WAL lets those -# readers run concurrently with the (serialized) writers. +# readers run concurrently with the (serialized) writers. The Postgres path keeps +# the same per-thread model (psycopg connections likewise aren't thread-safe). _local = threading.local() -def _conn() -> sqlite3.Connection | None: +def _open_pg(): + """Open a thread-local psycopg connection and ensure the schema. autocommit so + reads never leave an idle-in-transaction and each write lands immediately (the + shared put/get code still calls ``.commit()``, a harmless no-op under autocommit).""" + import psycopg # local import: only the Postgres path needs it + conn = psycopg.connect(_PG_DSN, autocommit=True) + for stmt in _PG_SCHEMA: + conn.execute(stmt) + return conn + + +def _conn(): conn = getattr(_local, "conn", None) if conn is not None: - return conn + # A dropped Postgres connection would otherwise disable the cache for the + # life of the thread; retire it so the next call reconnects. + if IS_POSTGRES and getattr(conn, "closed", False): + _local.conn = None + else: + return conn try: - os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) - conn = sqlite3.connect(DB_PATH, timeout=5.0) - conn.execute("PRAGMA journal_mode=WAL") - conn.execute("PRAGMA synchronous=NORMAL") - conn.executescript(_SCHEMA) + if IS_POSTGRES: + conn = _open_pg() + else: + os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) + conn = sqlite3.connect(DB_PATH, timeout=5.0) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA synchronous=NORMAL") + conn.executescript(_SCHEMA) _local.conn = conn return conn except Exception: # noqa: BLE001 - the store is optional; readers fall back @@ -77,13 +167,12 @@ def get_payload(kind: str, cell_id: str, key: str, token: str) -> bytes | None: if conn is None: return None try: - row = conn.execute( - "SELECT token, payload FROM derived WHERE kind=? AND cell_id=? AND key=?", - (kind, cell_id, key), - ).fetchone() + row = conn.execute(_SQL_GET_PAYLOAD, (kind, cell_id, key)).fetchone() if row is None or row[0] != token: return None - return zlib.decompress(row[1]) + # psycopg returns bytea as bytes/memoryview; bytes() normalizes both. + data = bytes(row[1]) if IS_POSTGRES else row[1] + return zlib.decompress(data) except Exception: # noqa: BLE001 return None @@ -111,7 +200,7 @@ def put_payload(kind: str, cell_id: str, key: str, token: str, payload: dict, if cache and conn is not None: try: conn.execute( - "INSERT OR REPLACE INTO derived VALUES (?,?,?,?,?,?)", + _SQL_PUT_PAYLOAD, (kind, cell_id, key, token, zlib.compress(body, 6), time.time()), ) conn.commit() @@ -138,7 +227,7 @@ def get_revgeo(key: str) -> tuple[bool, str | None]: if conn is None: return (False, None) try: - row = conn.execute("SELECT label, updated_at FROM revgeo WHERE key=?", (key,)).fetchone() + row = conn.execute(_SQL_GET_REVGEO, (key,)).fetchone() if row is None: return (False, None) if row[0] is None and time.time() - row[1] > REVGEO_MISS_TTL: @@ -153,7 +242,7 @@ def put_revgeo(key: str, label: str | None) -> None: if conn is None: return try: - conn.execute("INSERT OR REPLACE INTO revgeo VALUES (?,?,?)", (key, label, time.time())) + conn.execute(_SQL_PUT_REVGEO, (key, label, time.time())) conn.commit() except Exception: # noqa: BLE001 pass diff --git a/deploy/entrypoint.sh b/deploy/entrypoint.sh new file mode 100755 index 0000000..0f2d447 --- /dev/null +++ b/deploy/entrypoint.sh @@ -0,0 +1,26 @@ +#!/usr/bin/env bash +# Container entrypoint: bring the accounts schema up to head, then serve. +# +# Alembic runs before uvicorn so a fresh Postgres volume gets its tables. It is +# retried because `db` can still be finishing startup even though compose waits on +# its healthcheck — a cold volume's first connection may race the server becoming +# ready to accept the migration. +set -euo pipefail + +cd /app/backend + +max=10 +tries=0 +echo "==> Running database migrations (alembic upgrade head)" +until alembic upgrade head; do + tries=$((tries + 1)) + if [ "$tries" -ge "$max" ]; then + echo "!! alembic upgrade head failed after ${max} attempts; giving up" >&2 + exit 1 + fi + echo " migration attempt ${tries}/${max} failed; retrying in 3s..." >&2 + sleep 3 +done + +echo "==> Starting uvicorn on 0.0.0.0:${PORT:-8137} with ${WORKERS:-4} worker(s)" +exec uvicorn app:app --host 0.0.0.0 --port "${PORT:-8137}" --workers "${WORKERS:-4}" diff --git a/migrate_accounts_to_pg.py b/migrate_accounts_to_pg.py new file mode 100644 index 0000000..76ff632 --- /dev/null +++ b/migrate_accounts_to_pg.py @@ -0,0 +1,78 @@ +"""One-shot copy of the accounts database from SQLite into PostgreSQL. + +Run ONCE during the cutover, after ``alembic upgrade head`` has built the empty +Postgres schema. The copy goes through the mapped ORM tables, so SQLAlchemy's type +processors coerce the values automatically — CHAR(36) UUID strings → native +``uuid``, 0/1 → ``bool``, epoch floats stay floats, JSON stays JSON — which a raw +``pg_dump``/CSV path would need manual casts for. + +Login tokens (``access_token``) are deliberately NOT copied: they're ephemeral, so +skipping them just makes everyone re-login once at cutover and avoids the tz/uuid +fiddliness. Integer-PK sequences are reset afterwards, or the first post-migration +insert would collide. + +Usage: + THERMOGRAPH_DATABASE_URL=postgresql+psycopg://thermograph:PW@127.0.0.1:5432/thermograph \\ + python migrate_accounts_to_pg.py --sqlite /opt/thermograph/data/accounts.sqlite +""" +import argparse +import os +import sys + +from sqlalchemy import create_engine, insert, select, text + +from accounts import models + +# Parents before children (FK order); AccessToken is intentionally omitted. +TABLES = [ + models.User.__table__, + models.Subscription.__table__, + models.PushSubscription.__table__, + models.PendingDigest.__table__, + models.Notification.__table__, # references user + subscription +] +# Integer surrogate-PK tables whose sequence must be advanced past the copied max. +SEQUENCE_TABLES = ["subscription", "push_subscription", "pending_digest", "notification"] + + +def _pg_sync_url() -> str: + url = os.environ.get("THERMOGRAPH_DATABASE_URL", "").strip() + if not url.startswith("postgresql"): + sys.exit("THERMOGRAPH_DATABASE_URL must be a postgresql URL (the migration target)") + return url.replace("+asyncpg", "+psycopg") + + +def migrate(sqlite_path: str) -> None: + if not os.path.exists(sqlite_path): + sys.exit(f"source SQLite file not found: {sqlite_path}") + src = create_engine(f"sqlite:///{sqlite_path}", future=True) + dst = create_engine(_pg_sync_url(), future=True) + + counts = {} + with src.connect() as s, dst.begin() as d: + for table in TABLES: + rows = [dict(r._mapping) for r in s.execute(select(table))] + if rows: + d.execute(insert(table), rows) # explicit PKs preserved (FKs stay intact) + counts[table.name] = len(rows) + # Advance each identity sequence past the copied rows. + for name in SEQUENCE_TABLES: + d.execute(text( + f"SELECT setval(pg_get_serial_sequence('{name}', 'id'), " + f"COALESCE((SELECT MAX(id) FROM {name}), 1), (SELECT COUNT(*) > 0 FROM {name}))" + )) + + src.dispose() + dst.dispose() + print("Copied (access_token skipped — users re-login once):") + for name, n in counts.items(): + print(f" {name:18} {n}") + print("Sequences reset for:", ", ".join(SEQUENCE_TABLES)) + + +if __name__ == "__main__": + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--sqlite", default=os.environ.get("THERMOGRAPH_ACCOUNTS_DB") + or os.path.join(os.path.dirname(__file__), "..", "data", "accounts.sqlite"), + help="path to the source accounts.sqlite (default: data/accounts.sqlite)") + migrate(os.path.abspath(ap.parse_args().sqlite)) diff --git a/requirements.txt b/requirements.txt index 7cb9a77..079aa15 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,9 +3,15 @@ uvicorn[standard]==0.34.0 httpx==0.28.1 polars==1.42.1 numpy==2.2.1 -# Accounts + notification subscriptions (see backend/db.py, users.py, notify.py). +# Accounts + notification subscriptions (see accounts/db.py, users.py, notify.py). +# Postgres in prod/containers: asyncpg (async web/store) + psycopg (sync notifier); +# aiosqlite is kept for the SQLite fallback the test suite runs on. alembic manages +# the accounts schema. fastapi-users[sqlalchemy]==15.0.5 aiosqlite==0.22.1 +asyncpg==0.30.0 +psycopg[binary]==3.2.3 +alembic==1.14.0 # Web Push (VAPID) delivery of notifications (see backend/push.py). Pulls in # py-vapid, cryptography, and http-ece. pywebpush==2.0.0 diff --git a/tests/accounts/test_db_engines.py b/tests/accounts/test_db_engines.py new file mode 100644 index 0000000..8bc5224 --- /dev/null +++ b/tests/accounts/test_db_engines.py @@ -0,0 +1,40 @@ +"""The accounts DB engine layer: dialect selection and the read/read-write split. + +These run on the SQLite fallback (no THERMOGRAPH_DATABASE_URL in the test env); +the Postgres path (asyncpg RO/RW engines) is exercised by the docker-compose stack, +not the unit suite, so tests stay Postgres-free and fast.""" +import asyncio + +from sqlalchemy import select + +from accounts import db + + +def test_defaults_to_sqlite_when_no_database_url(): + assert db.IS_POSTGRES is False + # Both engines exist; on SQLite the RO engine is an alias of the RW one. + assert db.async_engine is not None + assert db.async_engine_ro is db.async_engine + + +def test_read_and_write_session_makers_are_distinct_dependencies(): + # Distinct FastAPI deps so pure-read endpoints can opt into the RO engine. + assert db.get_async_session is not db.get_read_session + assert callable(db.get_async_session) and callable(db.get_read_session) + + +def test_sync_url_derivation(): + assert db._sync_url("postgresql+asyncpg://u:p@h:5432/d") == "postgresql+psycopg://u:p@h:5432/d" + assert db._sync_url("sqlite+aiosqlite:////tmp/x.sqlite") == "sqlite:////tmp/x.sqlite" + + +def test_read_session_yields_a_working_session(): + # On SQLite the read session is fully usable (no read-only enforcement); this + # just confirms the dependency wiring resolves to a live AsyncSession. + db.Base.metadata.create_all(db.sync_engine) + + async def _use(): + async for session in db.get_read_session(): + return (await session.execute(select(1))).scalar_one() + + assert asyncio.run(_use()) == 1