| """Async database session helpers.""" |
|
|
| from collections.abc import AsyncGenerator |
|
|
| from sqlalchemy.ext.asyncio import ( |
| AsyncSession, |
| async_sessionmaker, |
| create_async_engine, |
| ) |
|
|
| from app.config import get_settings |
| from app.models import Base |
|
|
| settings = get_settings() |
|
|
| engine = create_async_engine( |
| settings.database_url, |
| echo=False, |
| ) |
|
|
| SessionLocal = async_sessionmaker( |
| engine, |
| expire_on_commit=False, |
| class_=AsyncSession, |
| ) |
|
|
|
|
| async def init_db() -> None: |
| """Create tables if they do not exist.""" |
| async with engine.begin() as conn: |
| await conn.run_sync(Base.metadata.create_all) |
|
|
|
|
| async def get_db() -> AsyncGenerator[AsyncSession, None]: |
| """Yield a database session.""" |
| async with SessionLocal() as session: |
| yield session |
|
|
