Chris TonjesandClaude Opus 5.5
0c359c926a
Before flipping control-mapping rows, wire the Phase 0 promises that had
no code behind them yet:
- SI-11 / C.5.7.43: app-wide exception handler returns
{"error":"internal_error","request_id":...} and the X-Request-ID header.
Logs exception type + stack frames, not the message — driver errors
echo bound parameters and key values. Engine set to hide_parameters.
- AU-11 / SI-12 (plan §7 risk): create_app refuses to start if any
mapped table lacks a retention policy.
- CM-3: migrations/IMMUTABLE.sha256 pins every merged revision; a test
fails on any edit or on an unpinned new revision.
- migrations/env.py: fileConfig(disable_existing_loggers=False) so
in-process Alembic (the drift test) doesn't silence app loggers.
163 tests passing; ruff, mypy, bandit, gitleaks clean.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
42 lines
1011 B
Python
42 lines
1011 B
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Iterator
|
|
|
|
from sqlalchemy import Engine, create_engine
|
|
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
|
|
|
|
from ypms.settings import get_settings
|
|
|
|
|
|
class Base(DeclarativeBase):
|
|
pass
|
|
|
|
|
|
_engine: Engine | None = None
|
|
_SessionLocal: sessionmaker[Session] | None = None
|
|
|
|
|
|
def engine() -> Engine:
|
|
global _engine
|
|
if _engine is None:
|
|
# hide_parameters: keep bound values (PII) out of exception text and logs.
|
|
_engine = create_engine(
|
|
get_settings().database_url, pool_pre_ping=True, future=True, hide_parameters=True
|
|
)
|
|
return _engine
|
|
|
|
|
|
def session_factory() -> sessionmaker[Session]:
|
|
global _SessionLocal
|
|
if _SessionLocal is None:
|
|
_SessionLocal = sessionmaker(bind=engine(), autoflush=False, autocommit=False, future=True)
|
|
return _SessionLocal
|
|
|
|
|
|
def get_db() -> Iterator[Session]:
|
|
db = session_factory()()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|