Sqlalchemy 2 Async

SQLAlchemy 2.0 Async Patterns ()

Modern async database patterns with SQLAlchemy 2.0, AsyncSession, and FastAPI integration.

Overview

• Building async FastAPI applications with database access
• Implementing async repository patterns
• Configuring async connection pooling
• Running concurrent database queries
• Avoiding N+1 queries in async context

Quick Reference

Engine and Session Factory

from sqlalchemy.ext.asyncio import (
    create_async_engine,
    async_sessionmaker,
    AsyncSession,
)

# Create async engine - ONE per application
engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,  # Verify connections before use
    pool_recycle=3600,   # Recycle connections after 1 hour
    echo=False,          # Set True for SQL logging in dev
)

# Session factory - use this to create sessions
async_session_factory = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,  # Prevent lazy load issues
    autoflush=False,         # Explicit flush control
)

FastAPI Dependency Injection

from typing import AsyncGenerator
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """Dependency that provides async database session."""
    async with async_session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

# Usage in route
@router.get("/users/{user_id}")
async def get_user(
    user_id: UUID,
    db: AsyncSession = Depends(get_db),
) -> UserResponse:
    result = await db.execute(
        select(User).where(User.id == user_id)
    )
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(404, "User not found")
    return UserResponse.model_validate(user)

Async Model Definition

from sqlalchemy import String, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy.dialects.postgresql import UUID
from datetime import datetime, timezone
import uuid

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        primary_key=True,
        default=uuid.uuid4,
    )
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    created_at: Mapped[datetime] = mapped_column(default=lambda: datetime.now(timezone.utc))

    # Relationship with explicit lazy loading strategy
    orders: Mapped[list["Order"]] = relationship(
        back_populates="user",
        lazy="raise",  # Prevent accidental lazy loads - MUST use selectinload
    )

class Order(Base):
    __tablename__ = "orders"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True)
    user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"))
    total: Mapped[int]

    user: Mapped["User"] = relationship(back_populates="orders", lazy="raise")

Eager Loading (Avoid N+1)

from sqlalchemy.orm import selectinload, joinedload
from sqlalchemy import select

async def get_user_with_orders(db: AsyncSession, user_id: UUID) -> User | None:
    """Load user with orders in single query - NO N+1."""
    result = await db.execute(
        select(User)
        .options(selectinload(User.orders))  # Eager load orders
        .where(User.id == user_id)
    )
    return result.scalar_one_or_none()

async def get_users_with_orders(db: AsyncSession, limit: int = 100) -> list[User]:
    """Load multiple users with orders efficiently."""
    result = await db.execute(
        select(User)
        .options(selectinload(User.orders))
        .limit(limit)
    )
    return list(result.scalars().all())

Bulk Operations ( Optimized)

async def bulk_insert_users(db: AsyncSession, users_data: list[dict]) -> int:
    """Efficient bulk insert - SQLAlchemy 2.0 uses multi-value INSERT."""
    # SQLAlchemy 2.0 automatically batches as single INSERT with multiple VALUES
    users = [User(**data) for data in users_data]
    db.add_all(users)
    await db.flush()  # Get IDs without committing
    return len(users)

async def bulk_insert_chunked(
    db: AsyncSession,
    items: list[dict],
    chunk_size: int = 1000,
) -> int:
    """Insert large datasets in chunks to manage memory."""
    total = 0
    for i in range(0, len(items), chunk_size):
        chunk = items[i:i + chunk_size]
        db.add_all([Item(**data) for data in chunk])
        await db.flush()
        total += len(chunk)
    return total

Repository Pattern

from typing import Generic, TypeVar
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

T = TypeVar("T", bound=Base)

class AsyncRepository(Generic[T]):
    """Generic async repository for CRUD operations."""

    def __init__(self, session: AsyncSession, model: type[T]):
        self.session = session
        self.model = model

    async def get(self, id: UUID) -> T | None:
        return await self.session.get(self.model, id)

    async def get_many(self, ids: list[UUID]) -> list[T]:
        result = await self.session.execute(
            select(self.model).where(self.model.id.in_(ids))
        )
        return list(result.scalars().all())

    async def create(self, **kwargs) -> T:
        instance = self.model(**kwargs)
        self.session.add(instance)
        await self.session.flush()
        return instance

    async def update(self, instance: T, **kwargs) -> T:
        for key, value in kwargs.items():
            setattr(instance, key, value)
        await self.session.flush()
        return instance

    async def delete(self, instance: T) -> None:
        await self.session.delete(instance)
        await self.session.flush()

Concurrent Queries with TaskGroup

import asyncio

async def get_dashboard_data(db: AsyncSession, user_id: UUID) -> dict:
    """Run multiple queries concurrently - same session is NOT thread-safe."""
    # WRONG: Don't share AsyncSession across tasks
    # async with asyncio.TaskGroup() as tg:
    #     tg.create_task(db.execute(...))  # NOT SAFE

    # CORRECT: Sequential queries with same session
    user = await db.get(User, user_id)
    orders_result = await db.execute(
        select(Order).where(Order.user_id == user_id).limit(10)
    )
    stats_result = await db.execute(
        select(func.count(Order.id)).where(Order.user_id == user_id)
    )

    return {
        "user": user,
        "recent_orders": list(orders_result.scalars().all()),
        "total_orders": stats_result.scalar(),
    }

async def get_data_from_multiple_users(user_ids: list[UUID]) -> list[dict]:
    """Concurrent queries - each task gets its own session."""
    async def fetch_user(user_id: UUID) -> dict:
        async with async_session_factory() as session:
            user = await session.get(User, user_id)
            return {"id": user_id, "email": user.email if user else None}

    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch_user(uid)) for uid in user_ids]

    return [t.result() for t in tasks]

Key Decisions

Anti-Patterns (FORBIDDEN)

# NEVER share AsyncSession across tasks
async with asyncio.TaskGroup() as tg:
    tg.create_task(session.execute(...))  # RACE CONDITION

# NEVER use sync Session in async code
from sqlalchemy.orm import Session
session = Session(engine)  # BLOCKS EVENT LOOP

# NEVER access lazy-loaded relationships without eager loading
user = await session.get(User)
orders = user.orders  # RAISES if lazy="raise", or BLOCKS if not

# NEVER use scoped_session with async
from sqlalchemy.orm import scoped_session
ScopedSession = scoped_session(session_factory)  # WRONG for async

# NEVER forget to handle session lifecycle
session = async_session_factory()
result = await session.execute(...)
# MISSING: session.close() - connection leak!

# NEVER use create_async_engine without pool_pre_ping in production
engine = create_async_engine(url)  # May use stale connections

Related Skills

• asyncio-advanced - TaskGroup and structured concurrency patterns
• alembic-migrations - Database migration with async support
• fastapi-advanced - Full FastAPI integration patterns
• database-schema-designer - Schema design best practices

Capability Details

async-session

Keywords: AsyncSession, async_sessionmaker, session factory, connection Solves:

• How do I create async database sessions?
• Configure async connection pooling
• Session lifecycle management

fastapi-integration

Keywords: Depends, dependency injection, get_db, request scope Solves:

• How do I integrate SQLAlchemy with FastAPI?
• Request-scoped database sessions
• Automatic commit/rollback handling

eager-loading

Keywords: selectinload, joinedload, eager load, N+1, relationship Solves:

• How do I avoid N+1 queries in async?
• Load relationships efficiently
• Configure lazy loading behavior

bulk-operations

Keywords: bulk insert, batch, chunk, add_all, performance Solves:

• How do I insert many rows efficiently?
• Chunk large inserts for memory
• SQLAlchemy 2.0 bulk optimizations

repository-pattern

Keywords: repository, CRUD, generic, base repository Solves:

• How do I implement repository pattern?
• Generic async CRUD operations
• Clean architecture with SQLAlchemy

SQLAlchemy 2.0 Async Checklist

Engine Configuration

• [ ] Using create_async_engine (not create_engine)
• [ ] Connection string uses async driver: postgresql+asyncpg://
• [ ] pool_pre_ping=True enabled for connection validation
• [ ] pool_size and max_overflow set appropriately
• [ ] pool_recycle set to prevent stale connections (e.g., 3600)

Session Factory

• [ ] Using async_sessionmaker (not sessionmaker)
• [ ] expire_on_commit=False to prevent lazy load issues
• [ ] autoflush=False for explicit control (optional)
• [ ] Single factory instance shared across application

FastAPI Integration

• [ ] Database dependency uses async with context manager
• [ ] Session yielded to routes, not returned
• [ ] Commit on success, rollback on exception
• [ ] Session properly closed after request

Model Definition

• [ ] Using Mapped[] type hints (SQLAlchemy 2.0 style)
• [ ] mapped_column() instead of Column()
• [ ] Relationships have explicit lazy= parameter
• [ ] lazy="raise" to prevent accidental lazy loads

Eager Loading

• [ ] Using selectinload() for collections
• [ ] Using joinedload() for single relationships
• [ ] All needed relationships loaded in query
• [ ] No N+1 queries in response serialization

Bulk Operations

• [ ] Using add_all() for multiple inserts
• [ ] Chunking large inserts (1000-10000 per batch)
• [ ] Using flush() between chunks for memory
• [ ] Batch size tuned for performance

Concurrency

• [ ] One AsyncSession per task/request (never shared)
• [ ] Not using scoped_session with async
• [ ] Concurrent queries use separate sessions
• [ ] Connection pool sized for concurrent load

Error Handling

• [ ] Proper exception handling around DB operations
• [ ] Rollback on errors before re-raising
• [ ] Connection errors handled gracefully
• [ ] Retry logic for transient failures

Testing

• [ ] Using test database (not production)
• [ ] Transactions rolled back after each test
• [ ] Async test fixtures with pytest-asyncio
• [ ] Database state isolated between tests

Performance

• [ ] Indexes on frequently queried columns
• [ ] EXPLAIN ANALYZE run on slow queries
• [ ] Connection pool metrics monitored
• [ ] Query execution time logged

SQLAlchemy 2.0 Async Examples

Example 1: Complete FastAPI Setup

# app/db/engine.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"

engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
    pool_recycle=3600,
    echo=False,
)

async_session_factory = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


# app/api/deps.py
from typing import AsyncGenerator

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise


# app/api/routes/users.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.orm import selectinload

router = APIRouter()

@router.get("/users/{user_id}")
async def get_user(user_id: UUID, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        select(User)
        .options(selectinload(User.orders))
        .where(User.id == user_id)
    )
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(404, "User not found")
    return user

Example 2: Model with Proper Type Hints

from datetime import datetime, timezone
from uuid import UUID, uuid4
from sqlalchemy import String, ForeignKey, DateTime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy.dialects.postgresql import UUID as PG_UUID

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[UUID] = mapped_column(
        PG_UUID(as_uuid=True),
        primary_key=True,
        default=uuid4,
    )
    email: Mapped[str] = mapped_column(
        String(255),
        unique=True,
        index=True,
    )
    name: Mapped[str] = mapped_column(String(100))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
    )

    # Prevent accidental lazy loading
    orders: Mapped[list["Order"]] = relationship(
        back_populates="user",
        lazy="raise",
    )

class Order(Base):
    __tablename__ = "orders"

    id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), primary_key=True)
    user_id: Mapped[UUID] = mapped_column(ForeignKey("users.id"))
    total_cents: Mapped[int]
    status: Mapped[str] = mapped_column(String(20), default="pending")

    user: Mapped["User"] = relationship(back_populates="orders", lazy="raise")
    items: Mapped[list["OrderItem"]] = relationship(lazy="raise")

Example 3: Repository Pattern

from typing import Generic, TypeVar
from uuid import UUID
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

T = TypeVar("T", bound=Base)

class AsyncRepository(Generic[T]):
    def __init__(self, session: AsyncSession, model: type[T]):
        self.session = session
        self.model = model

    async def get(self, id: UUID) -> T | None:
        return await self.session.get(self.model, id)

    async def get_by_ids(self, ids: list[UUID]) -> list[T]:
        if not ids:
            return []
        result = await self.session.execute(
            select(self.model).where(self.model.id.in_(ids))
        )
        return list(result.scalars().all())

    async def list(
        self,
        *,
        offset: int = 0,
        limit: int = 100,
    ) -> list[T]:
        result = await self.session.execute(
            select(self.model).offset(offset).limit(limit)
        )
        return list(result.scalars().all())

    async def create(self, **kwargs) -> T:
        instance = self.model(**kwargs)
        self.session.add(instance)
        await self.session.flush()
        return instance

    async def update(self, instance: T, **kwargs) -> T:
        for key, value in kwargs.items():
            setattr(instance, key, value)
        await self.session.flush()
        return instance

    async def delete(self, instance: T) -> None:
        await self.session.delete(instance)
        await self.session.flush()


# Usage
class UserRepository(AsyncRepository[User]):
    def __init__(self, session: AsyncSession):
        super().__init__(session, User)

    async def get_by_email(self, email: str) -> User | None:
        result = await self.session.execute(
            select(User).where(User.email == email)
        )
        return result.scalar_one_or_none()

    async def get_with_orders(self, user_id: UUID) -> User | None:
        result = await self.session.execute(
            select(User)
            .options(selectinload(User.orders))
            .where(User.id == user_id)
        )
        return result.scalar_one_or_none()

Example 4: Bulk Operations

async def bulk_create_users(
    db: AsyncSession,
    users_data: list[dict],
    chunk_size: int = 1000,
) -> int:
    """Efficiently insert many users in chunks."""
    total = 0

    for i in range(0, len(users_data), chunk_size):
        chunk = users_data[i:i + chunk_size]
        users = [User(**data) for data in chunk]
        db.add_all(users)
        await db.flush()  # Get IDs, manage memory
        total += len(chunk)

    return total


async def bulk_update_status(
    db: AsyncSession,
    order_ids: list[UUID],
    new_status: str,
) -> int:
    """Bulk update using UPDATE statement."""
    from sqlalchemy import update

    result = await db.execute(
        update(Order)
        .where(Order.id.in_(order_ids))
        .values(status=new_status)
    )
    return result.rowcount

Example 5: Transaction Management

from sqlalchemy.ext.asyncio import AsyncSession

async def transfer_funds(
    db: AsyncSession,
    from_account_id: UUID,
    to_account_id: UUID,
    amount: int,
) -> None:
    """Transfer with explicit transaction and row locking."""
    async with db.begin():  # Explicit transaction
        # Lock rows to prevent concurrent modification
        from_account = await db.get(
            Account,
            from_account_id,
            with_for_update=True,
        )
        to_account = await db.get(
            Account,
            to_account_id,
            with_for_update=True,
        )

        if not from_account or not to_account:
            raise ValueError("Account not found")

        if from_account.balance < amount:
            raise ValueError("Insufficient funds")

        from_account.balance -= amount
        to_account.balance += amount

        # Transaction commits on exit, rolls back on exception

Example 6: Complex Queries with Joins

from sqlalchemy import select, func, and_
from sqlalchemy.orm import selectinload, joinedload

async def get_user_order_summary(
    db: AsyncSession,
    user_id: UUID,
) -> dict:
    """Get user with order statistics."""
    # Get user with eager-loaded orders
    user_result = await db.execute(
        select(User)
        .options(selectinload(User.orders))
        .where(User.id == user_id)
    )
    user = user_result.scalar_one_or_none()

    if not user:
        return None

    # Get aggregate stats
    stats_result = await db.execute(
        select(
            func.count(Order.id).label("total_orders"),
            func.sum(Order.total_cents).label("total_spent"),
            func.avg(Order.total_cents).label("avg_order"),
        )
        .where(Order.user_id == user_id)
    )
    stats = stats_result.one()

    return {
        "user": user,
        "total_orders": stats.total_orders,
        "total_spent_cents": stats.total_spent or 0,
        "avg_order_cents": float(stats.avg_order or 0),
    }

Eager Loading Patterns for Async SQLAlchemy

The N+1 Problem in Async

# BAD: N+1 queries - one for users, N for orders
async def get_users_bad(db: AsyncSession) -> list[User]:
    result = await db.execute(select(User))
    users = result.scalars().all()
    for user in users:
        # This triggers N additional queries (or raises if lazy="raise")
        print(user.orders)
    return users

# GOOD: Single query with eager loading
async def get_users_good(db: AsyncSession) -> list[User]:
    result = await db.execute(
        select(User).options(selectinload(User.orders))
    )
    users = result.scalars().all()
    for user in users:
        print(user.orders)  # Already loaded
    return users

Loading Strategies

selectinload (Recommended for Collections)

from sqlalchemy.orm import selectinload

# Loads orders in separate SELECT ... WHERE user_id IN (...)
result = await db.execute(
    select(User)
    .options(selectinload(User.orders))
    .limit(100)
)

joinedload (Best for Single Relations)

from sqlalchemy.orm import joinedload

# Uses LEFT JOIN - good for to-one relationships
result = await db.execute(
    select(Order)
    .options(joinedload(Order.user))
    .where(Order.status == "pending")
)

Nested Eager Loading

# Load user -> orders -> order_items
result = await db.execute(
    select(User)
    .options(
        selectinload(User.orders).selectinload(Order.items)
    )
)

# Load user -> orders and user -> addresses
result = await db.execute(
    select(User)
    .options(
        selectinload(User.orders),
        selectinload(User.addresses),
    )
)

Configuring Models to Prevent Lazy Load

from sqlalchemy.orm import relationship, Mapped

class User(Base):
    __tablename__ = "users"

    id: Mapped[UUID] = mapped_column(primary_key=True)

    # lazy="raise" prevents accidental lazy loading
    # Forces explicit eager loading
    orders: Mapped[list["Order"]] = relationship(
        back_populates="user",
        lazy="raise",  # Raises if accessed without eager load
    )

    # For optional relationships you might want loaded
    profile: Mapped["Profile"] = relationship(
        lazy="joined",  # Always joined (use sparingly)
    )

Strategy Comparison

Dynamic Loading for Large Collections

class User(Base):
    # For very large collections, use dynamic loading
    orders: Mapped[list["Order"]] = relationship(
        lazy="dynamic",  # Returns query, not collection
    )

# Usage
async def get_recent_orders(db: AsyncSession, user_id: UUID) -> list[Order]:
    user = await db.get(User, user_id)
    # Dynamic relationship returns a query
    result = await db.execute(
        user.orders.limit(10).order_by(Order.created_at.desc())
    )
    return list(result.scalars().all())

FastAPI + SQLAlchemy 2.0 Async Integration

Complete Setup

# app/db/session.py
from sqlalchemy.ext.asyncio import (
    create_async_engine,
    async_sessionmaker,
    AsyncSession,
)
from app.core.config import settings

engine = create_async_engine(
    settings.DATABASE_URL,
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
    pool_recycle=3600,
    echo=settings.DEBUG,
)

async_session_factory = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autoflush=False,
)

Dependency Injection

# app/api/deps.py
from typing import AsyncGenerator
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.session import async_session_factory

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """Provide database session with automatic cleanup."""
    async with async_session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

Route with Database Access

# app/api/v1/routes/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.orm import selectinload
from sqlalchemy.ext.asyncio import AsyncSession

from app.api.deps import get_db
from app.models.user import User
from app.schemas.user import UserResponse, UserCreate

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: UUID,
    db: AsyncSession = Depends(get_db),
) -> UserResponse:
    result = await db.execute(
        select(User)
        .options(selectinload(User.orders))
        .where(User.id == user_id)
    )
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "User not found")
    return UserResponse.model_validate(user)

@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_in: UserCreate,
    db: AsyncSession = Depends(get_db),
) -> UserResponse:
    user = User(**user_in.model_dump())
    db.add(user)
    await db.flush()  # Get ID without committing
    await db.refresh(user)  # Load any defaults
    return UserResponse.model_validate(user)

Service Layer Pattern

# app/services/user_service.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate

class UserService:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get(self, user_id: UUID) -> User | None:
        return await self.db.get(User, user_id)

    async def get_by_email(self, email: str) -> User | None:
        result = await self.db.execute(
            select(User).where(User.email == email)
        )
        return result.scalar_one_or_none()

    async def create(self, user_in: UserCreate) -> User:
        user = User(**user_in.model_dump())
        self.db.add(user)
        await self.db.flush()
        return user

    async def update(self, user: User, user_in: UserUpdate) -> User:
        for field, value in user_in.model_dump(exclude_unset=True).items():
            setattr(user, field, value)
        await self.db.flush()
        return user

# Usage in route
@router.post("/")
async def create_user(
    user_in: UserCreate,
    db: AsyncSession = Depends(get_db),
):
    service = UserService(db)
    if await service.get_by_email(user_in.email):
        raise HTTPException(400, "Email already registered")
    return await service.create(user_in)

Transaction Management

# Explicit transaction control
@router.post("/transfer")
async def transfer_funds(
    transfer: TransferRequest,
    db: AsyncSession = Depends(get_db),
):
    async with db.begin():  # Explicit transaction
        from_account = await db.get(Account, transfer.from_id, with_for_update=True)
        to_account = await db.get(Account, transfer.to_id, with_for_update=True)

        if from_account.balance < transfer.amount:
            raise HTTPException(400, "Insufficient funds")

        from_account.balance -= transfer.amount
        to_account.balance += transfer.amount
        # Commits automatically on exit, rolls back on exception

Lifespan with Database

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.db.session import engine

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: verify database connection
    async with engine.begin() as conn:
        await conn.execute(text("SELECT 1"))

    yield

    # Shutdown: dispose engine
    await engine.dispose()

app = FastAPI(lifespan=lifespan)