비동기 DB (asyncpg + AsyncSession)

create_async_engine async_sessionmaker AsyncSession await greenlet

동기 버전과 구조는 같고, 엔진·세션·쿼리에 async/await 가 붙습니다.

1. 설치

pip install "sqlalchemy[asyncio]>=2.0" asyncpg

2. 비동기 엔진·세션

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

engine = create_async_engine(
    "postgresql+asyncpg://school:secret@localhost:5432/school",
    pool_size=5, max_overflow=10, pool_pre_ping=True,
)

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

3. 의존성

from typing import Annotated
from fastapi import Depends

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

AsyncDbDep = Annotated[AsyncSession, Depends(get_db)]

4. 쿼리

from sqlalchemy import select

@app.get("/posts")
async def list_posts(db: AsyncDbDep, limit: int = 20):
    result = await db.scalars(select(Post).limit(limit))
    return result.all()

@app.post("/posts")
async def create_post(payload: PostIn, db: AsyncDbDep):
    post = Post(**payload.model_dump())
    db.add(post)
    await db.commit()
    await db.refresh(post)
    return post

5. 지연 로딩(lazy loading) 주의

async 에서는 lazy load 불가

post.author 처럼 관계를 나중에 접근하면 MissingGreenlet 에러
해결: select(Post).options(selectinload(Post.author)) 로 eager 로딩
또는 lazy="selectin" 을 relationship 기본값으로

from sqlalchemy.orm import selectinload

stmt = select(Post).options(selectinload(Post.author)).limit(20)
posts = (await db.scalars(stmt)).all()

6. 테이블 생성

async def init_models():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

"마이그레이션 없는 스키마 변경은 사고의 시작이다"