"""Async SQLAlchemy engine/session management (SPEC ยง2 ``app/db.py``).""" from __future__ import annotations from collections.abc import AsyncIterator from contextlib import asynccontextmanager from sqlalchemy.ext.asyncio import ( AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine, ) from sqlalchemy.orm import DeclarativeBase from app.config import get_settings _engine: AsyncEngine | None = None _sessionmaker: async_sessionmaker[AsyncSession] | None = None class Base(DeclarativeBase): """Declarative base for all ORM models.""" def get_engine() -> AsyncEngine: """Return (creating on first use) the process-wide async engine.""" global _engine if _engine is None: _engine = create_async_engine( get_settings().database_url, pool_pre_ping=True, future=True, ) return _engine def get_sessionmaker() -> async_sessionmaker[AsyncSession]: """Return the process-wide session factory.""" global _sessionmaker if _sessionmaker is None: _sessionmaker = async_sessionmaker( get_engine(), expire_on_commit=False, class_=AsyncSession ) return _sessionmaker @asynccontextmanager async def session_scope() -> AsyncIterator[AsyncSession]: """Transactional session context manager for workers and scripts.""" async with get_sessionmaker()() as session: try: yield session await session.commit() except Exception: await session.rollback() raise async def get_session() -> AsyncIterator[AsyncSession]: """FastAPI dependency yielding a request-scoped session.""" async with get_sessionmaker()() as session: yield session async def dispose_engine() -> None: """Dispose the engine (used on app shutdown).""" global _engine, _sessionmaker if _engine is not None: await _engine.dispose() _engine = None _sessionmaker = None