"""DataK3 connection + the one write helper every route uses.
A DataK3 bucket is a Postgres endpoint: db name = the bucket, user = the literal
``token``, password = your DODIL login (or service-account) token. Endpoints are fixed
region constants — nothing to discover. See .env.example.
"""
import os
import psycopg
from sqlalchemy import event
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import sessionmaker
from erp_platform.pool import pooled_engine
from erp_platform.sa_token import get_pg_password
BUCKET = os.environ["BUCKET"] # no default: a silent default writes to somebody else's bucket
PG_HOST = os.environ.get("PG_HOST", "pg.uk-lon-1.dodil.io")
PG_PORT = int(os.environ.get("PG_PORT", "5432"))
# Which environment this app believes it is. The bucket carries the same claim in a row, and
# the two are compared on the first query — see assert_environment below.
EXPECT_ENV = os.environ.get("DODIL_ENV", "").strip().lower()
# sslmode=require: the pg wire has TLS (gateway v1.0.5+), and the password on it is a
# service-account secret — never let a driver fall back to plaintext. verify-full is the
# stronger posture, but libpq does not read the OS trust store, so it needs an explicit
# sslrootcert (see PLATFORM.md). The password is resolved lazily per-connection so nothing reads
# a token at import (openapi() builds with no creds), and a deployed app mints a fresh
# service-account token before it expires.
def _connect():
return psycopg.connect(host=PG_HOST, port=PG_PORT, dbname=BUCKET,
user="token", password=get_pg_password(), sslmode="require")
# ── which bucket is this, really ───────────────────────────────────────────────────────────
# A credential cannot answer that question. No role in k3-authorization-service is
# conditionable (verified 2026-09-18), so a token that writes one bucket writes EVERY bucket in
# the org: permissions cannot stop an app pointed at the wrong one. The only thing that can is
# the app, and the only claim it can trust is one the BUCKET makes about itself — an env var
# saying "this is dev" is exactly the thing that goes missing.
#
# So a bucket is STAMPED once (`python migrate.py --stamp <env>`) and every app declares the
# environment it expects (DODIL_ENV). They are compared on the first query.
#
# NOT at import, and NOT in /healthz. A pod must report healthy before its credentials resolve
# or it never starts, so this runs on first use: /healthz keeps answering, the self-check says
# exactly what is wrong, and every query fails loudly instead of writing into someone's books.
_env_checked = False
class WrongBucket(RuntimeError):
"""Raised when the bucket disagrees with the environment this app was deployed as."""
def assert_environment(conn=None) -> dict:
"""Compare this app's declared environment with the bucket's own stamp.
Unstamped buckets: an app expecting `prod` refuses, because the one place this must not
fail open is the one holding a customer's books. Anything else warns — most scratch buckets
predate this and failing them closed would only teach people to unset DODIL_ENV.
"""
global _env_checked
own = conn is None
if own:
conn = engine.connect()
try:
try:
row = conn.exec_driver_sql(
"SELECT value FROM platform_meta WHERE key = 'environment'").fetchone()
except Exception: # noqa: BLE001 — no table = unstamped
row = None
actual = (row[0] if row else "").strip().lower()
if EXPECT_ENV and actual and EXPECT_ENV != actual:
raise WrongBucket(
f"refusing to use bucket {BUCKET!r}: it is stamped {actual!r} and this app was "
f"deployed as {EXPECT_ENV!r}. Nothing has been read or written. Fix DODIL_ENV or "
f"BUCKET — do not stamp the bucket to make this go away.")
if EXPECT_ENV == "prod" and not actual:
raise WrongBucket(
f"refusing to use bucket {BUCKET!r}: this app is deployed as 'prod' and the "
f"bucket carries no environment stamp. Stamp it with "
f"`python migrate.py --stamp prod` once you have confirmed it is the right one.")
_env_checked = True
return {"bucket": BUCKET, "stamped": actual or None, "expected": EXPECT_ENV or None,
"warning": (f"bucket {BUCKET!r} is not stamped" if EXPECT_ENV and not actual
else None)}
finally:
if own:
conn.close()
# expire_on_commit=False is deliberate: DataK3 has NO read-your-writes inside an open
# transaction (the write-log is staged until commit), so we never re-read an instance's
# columns mid-transaction — we keep the values we committed.
# DB_CONCURRENCY caps this process's connections — an admission limit BELOW the tables reader's hard
# 16-concurrent-read serve cap (ConfigurationLimitExceeded). A 60-way storm went from lossy to 60/60
# clean with a cap of 10 plus a retry (CLAUDE.md, verified 2026-09-03). Waiters queue for a connection
# (pool_timeout) instead of overrunning the reader; erp_platform.retry absorbs what still gets through.
DB_CONCURRENCY = int(os.environ.get("DB_CONCURRENCY", "8"))
# 0.9.6: the pool lives in erp_platform.pool — LIFO, pinged only after DB_PING_AFTER idle seconds (a ping is a round
# trip on every checkout otherwise), recycled after DB_POOL_RECYCLE; each new connection resolves a current token.
engine = pooled_engine(_connect, pool_size=DB_CONCURRENCY, max_overflow=0, pool_timeout=30)
Session = sessionmaker(engine, expire_on_commit=False)
@event.listens_for(engine, "connect")
def _check_environment_once(dbapi_conn, _record):
"""First real connection, and only the first: is this the bucket we were deployed against?
On the connect event rather than at import, so nothing reads a token while the module loads
and /healthz can answer before credentials resolve. A mismatch raises here, which fails the
connection — every query then fails with the reason, and not one of them touched a row.
"""
global _env_checked
if _env_checked or not EXPECT_ENV:
return
cur = dbapi_conn.cursor()
try:
cur.execute("SELECT value FROM platform_meta WHERE key = 'environment'")
row = cur.fetchone()
actual = (row[0] if row else "").strip().lower()
except Exception: # noqa: BLE001 — no table = unstamped
dbapi_conn.rollback()
actual = ""
finally:
cur.close()
if actual and EXPECT_ENV != actual:
raise WrongBucket(
f"refusing to use bucket {BUCKET!r}: it is stamped {actual!r} and this app was "
f"deployed as {EXPECT_ENV!r}. Nothing has been read or written.")
if EXPECT_ENV == "prod" and not actual:
raise WrongBucket(
f"refusing to use bucket {BUCKET!r}: deployed as 'prod' against a bucket carrying "
f"no environment stamp.")
_env_checked = True
def upsert(session, model, rows, key, insert_only=("created_at", "created_by")):
"""Idempotent keyed write — ``INSERT ... ON CONFLICT (<key>) DO UPDATE``.
On DataK3 a bare re-INSERT of an already-committed primary key raises a duplicate-key
error (SQLSTATE 23505) — a plain INSERT is not an upsert on re-write. This is the
idempotent form: re-running (a retry, a shard replay, a re-import) lands the row once.
``rows`` is a list of dicts; ``key`` is the primary-key column name(s).
Delegates to repository.upsert, which fixed two defects in the form that lived here:
a batched row could have a column it never named set to NULL (the batch was padded to a
common column set with None, then every padded column was updated), and created_at /
created_by were overwritten on every re-write. See repository.upsert.
"""
from erp_platform.repository import upsert as _upsert
_upsert(session, model, rows, key, insert_only)