Chris TonjesandClaude Opus 4.7
393631ad69
Scaffolding for the YPMS in-house build. Mirrors OWC's structure so future OWC framework code can be vendored in with minimal fiddling (milestone 0.2). Includes: - pyproject.toml — Python 3.12, FastAPI, SQLAlchemy 2.x, Alembic, Postgres, pydantic-settings, structlog, cryptography. Dev extras: pytest, ruff, mypy, bandit, pip-audit. - Dockerfile — multi-stage builder + slim runtime. - docker-compose.yml — app + Postgres (on 5433 to avoid clashing with OWC on 5432) + Adminer (on 8081). - alembic.ini + migrations/env.py + baseline 0001_baseline.py (empty by design; every future change is additive and immutable once merged). - src/ypms/main.py — FastAPI factory with /healthz and /readyz. - src/ypms/settings.py — pydantic-settings config (DB URL, JWT, column encryption key, log level, env, CORS). - src/ypms/logging.py — structlog + sanitize_for_log() for the sensitive-fields-never-in-logs rule from ARCHITECTURE.md §10.1. - src/ypms/db.py — SQLAlchemy Base, engine, session factory, get_db. - src/ypms/middleware.py — RequestIdMiddleware; X-Request-ID header echoed both ways. - src/ypms/framework/ and src/ypms/security/ — empty packages, to be populated in milestones 0.2 and 0.3-0.4. - tests/test_health.py — 3 smoke tests (healthz OK, header present, header honoured on incoming). - README.md — quickstart. - CONTRIBUTING.md — hard rules mirrored from OWC. - .github/workflows/ci.yml — lint (ruff), types (mypy), tests + 80% coverage floor, SAST (bandit), SCA (pip-audit), gitleaks. Green locally: ruff clean, mypy clean, pytest 3 passed. Advances YPMS-POAM-040 (architecture doc + Phase 0 kickoff). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
39 lines
884 B
Python
39 lines
884 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:
|
|
_engine = create_engine(get_settings().database_url, pool_pre_ping=True, future=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()
|