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>
57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
"""Alembic environment.
|
|
|
|
Model modules are imported here so `Base.metadata` is populated before Alembic
|
|
compares against it. New model modules go in `src/ypms/models.py`, not here.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from logging.config import fileConfig
|
|
|
|
from alembic import context
|
|
from sqlalchemy import engine_from_config, pool
|
|
|
|
from ypms.db import Base
|
|
from ypms.settings import get_settings
|
|
|
|
# ---- Populate Base.metadata ----
|
|
import ypms.models # noqa: F401 — every model module; see src/ypms/models.py
|
|
|
|
config = context.config
|
|
if config.config_file_name is not None:
|
|
# Don't disable loggers configured before Alembic runs (app, or pytest when
|
|
# the drift test invokes migrations in-process).
|
|
fileConfig(config.config_file_name, disable_existing_loggers=False)
|
|
|
|
config.set_main_option("sqlalchemy.url", get_settings().database_url)
|
|
target_metadata = Base.metadata
|
|
|
|
|
|
def run_migrations_offline() -> None:
|
|
context.configure(
|
|
url=config.get_main_option("sqlalchemy.url"),
|
|
target_metadata=target_metadata,
|
|
literal_binds=True,
|
|
dialect_opts={"paramstyle": "named"},
|
|
)
|
|
with context.begin_transaction():
|
|
context.run_migrations()
|
|
|
|
|
|
def run_migrations_online() -> None:
|
|
connectable = engine_from_config(
|
|
config.get_section(config.config_ini_section, {}),
|
|
prefix="sqlalchemy.",
|
|
poolclass=pool.NullPool,
|
|
)
|
|
with connectable.connect() as connection:
|
|
context.configure(connection=connection, target_metadata=target_metadata)
|
|
with context.begin_transaction():
|
|
context.run_migrations()
|
|
|
|
|
|
if context.is_offline_mode():
|
|
run_migrations_offline()
|
|
else:
|
|
run_migrations_online()
|