""" Database configuration and session management using SQLAlchemy async. """ from typing import AsyncGenerator from sqlalchemy import MetaData from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from app.config import settings # Create async engine engine = create_async_engine( settings.DATABASE_URL, echo=settings.DEBUG, future=True, ) # Create async session factory async_session_maker = sessionmaker( engine, class_=AsyncSession, expire_on_commit=False, autocommit=False, autoflush=False, ) # Naming convention for constraints (for Alembic) convention = { "ix": "ix_%(column_0_label)s", "uq": "uq_%(table_name)s_%(column_0_name)s", "ck": "ck_%(table_name)s_%(constraint_name)s", "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s", "pk": "pk_%(table_name)s", } metadata = MetaData(naming_convention=convention) # Base class for all models Base = declarative_base(metadata=metadata) async def get_db() -> AsyncGenerator[AsyncSession, None]: """ Dependency to get database session. Usage: @app.get("/users") async def get_users(db: AsyncSession = Depends(get_db)): ... Yields: AsyncSession instance """ async with async_session_maker() as session: try: yield session finally: await session.close()