"""The pooled DataK3 engine every helper builds on (erp_platform.db, erp_platform.livecheck).
Connect + auth costs ~1.2 s on the pg wire and a statement ~70-100 ms (measured from a Mac to pg.uk-lon-1,
2026-10-24), so an authenticated connection is worth keeping and a ping is worth skipping when it is not
needed:
* A QueuePool of `pool_size` (DB_CONCURRENCY in erp_platform.db, an admission cap below the reader's 16), LIFO,
so the few hot connections are the ones reused and the idle ones age out.
* PRE-PING ONLY AFTER IDLE. SQLAlchemy's `pool_pre_ping` sends a `SELECT 1` on EVERY checkout — one more round
trip per request. Here a connection is pinged when it sat in the pool longer than DB_PING_AFTER seconds
(default 10; 0 pings on every checkout, the classic pre-ping). A dead one is replaced before the use case
sees it, so a server-side close never reaches a command as the ambiguous 08006 (erp_platform.retry).
* RECYCLE after DB_POOL_RECYCLE seconds (default 1800). The password is resolved by `creator` for every NEW
connection — erp_platform.sa_token mints and caches a fresh service-account token before expiry — so a
recycled or replaced connection authenticates with a current token; an open one is already authenticated.
"""
import os
import time
from typing import Callable
from sqlalchemy import create_engine, event, exc
PING_AFTER = float(os.environ.get("DB_PING_AFTER", "10"))
RECYCLE = int(os.environ.get("DB_POOL_RECYCLE", "1800"))
_IDLE = "erp_idle_since"
def pooled_engine(creator: Callable, *, pool_size: int, max_overflow: int = 0, pool_timeout: float = 30,
ping_after: float = PING_AFTER, recycle: int = RECYCLE):
# 0.9.11: use_native_hstore=False — psycopg's dialect otherwise looks the hstore type up in pg_catalog on the
# first connection, and on DataK3 every pg_catalog query costs O(tables) today (dodilio/tables#494: ~178 s at
# 288 tables). The small-ERP deploy's first connection went from 6 min to 2 s with it off; no brick uses hstore.
e = create_engine("postgresql+psycopg://", creator=creator, pool_size=pool_size, max_overflow=max_overflow,
pool_timeout=pool_timeout, pool_recycle=recycle, pool_use_lifo=True,
pool_pre_ping=False, future=True, use_native_hstore=False)
watch_idle(e, ping_after)
return e
def watch_idle(engine, ping_after: float = PING_AFTER) -> None:
"""Ping a pooled connection on checkout when it has been idle longer than `ping_after` seconds."""
@event.listens_for(engine, "checkin")
def _idle(_dbapi_conn, record):
if record is not None:
record.info[_IDLE] = time.monotonic()
@event.listens_for(engine, "checkout")
def _ping(dbapi_conn, record, _proxy):
since = record.info.pop(_IDLE, None)
if since is None or time.monotonic() - since < ping_after:
return
try:
alive = engine.dialect.do_ping(dbapi_conn)
except Exception as e: # noqa: BLE001 — any failure: replace it
raise exc.DisconnectionError(f"pooled connection failed its ping after idle: {type(e).__name__}") from e
if alive is False:
raise exc.DisconnectionError("pooled connection failed its ping after idle")