25 장

DB 연동 — SQLAlchemy 2.x + Alembic

SQLAlchemy 2.x의 새 스타일 — Mapped/mapped_column, async 세션, FastAPI 의존성 주입과의 결합, Alembic 마이그레이션까지 정리합니다.

23장 라우팅, Pydantic 모델, 의존성 주입의 인메모리 dict를 본 챕터에서 DB로 교체 합니다. SQLAlchemy 2.x의 새 스타일, async 세션, FastAPI 의존성 주입과의 결합, 그리고 스키마 변경을 안전하게 다루는 Alembic 마이그레이션까지입니다.

본 챕터의 Mapped[T] 표기와 ORM 매핑은 16장 디스크립터와 __set_name__의 디스크립터, 17장 메타클래스의 ORM DSL 패턴이 결합된 결과입니다. 비동기 세션은 14장 비동기 입문 / 18장 비동기 깊이 위에 올라갑니다.

왜 SQLAlchemy 2.x 인가 #

Python의 ORM은 여럿이지만 (SQLAlchemy, Tortoise, SQLModel, Peewee, Django ORM), 모던 비동기 + 타입 친화 관점에서 가장 표준 위치에 있는 것은 SQLAlchemy 2.x입니다.

2.x는 1.x와 같은 라이브러리지만 사용 스타일이 사실상 달라졌습니다.

1.x 스타일	2.x 스타일
`Column(String, ...)`	`Mapped[str] = mapped_column(...)`
`session.query(User).filter(...)`	`session.execute(select(User).where(...))`
동기만 (3rd party 비동기)	빌트인 async
타입 힌트 보조적	타입 힌트 핵심

새 코드는 무조건 2.x 스타일입니다.

SQLModel은? #

FastAPI를 만든 사람이 설계한 라이브러리로, SQLAlchemy + Pydantic 통합을 시도한 것입니다. 짧은 코드에는 유리하지만, 아직 SQLAlchemy 2.x의 모든 기능을 노출하지 않아 복잡한 쿼리에서는 한계가 있습니다. 새 프로젝트는 SQLAlchemy 2.x 직접 + Pydantic 별도를 권장합니다.

설치 #

DB 의존성

uv add sqlalchemy "sqlalchemy[asyncio]" asyncpg
uv add --dev alembic

sqlalchemy — ORM 본체
asyncpg — PostgreSQL 비동기 드라이버 (async 사용 시)
alembic — 마이그레이션 도구

SQLite로 시작하려면 aiosqlite (PostgreSQL 대신).

SQLite로 빠르게

uv add aiosqlite

Engine과 세션 #

app/db/session.py

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine, AsyncSession
from app.core.config import settings

engine = create_async_engine(
    settings.database_url,
    echo=False,           # 개발 중에는 True 로 SQL 로그
    pool_pre_ping=True,   # 끊어진 연결 자동 복구
)

AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,   # 커밋 후에도 객체 사용 가능
)

expire_on_commit=False는 비동기 환경에서 거의 필수. True 면 커밋 후 속성 접근이 lazy load를 트리거하는데, async 함수 밖에서는 위험합니다.

의존성으로 세션 주입 #

app/api/deps.py

from typing import Annotated, AsyncIterator
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.session import AsyncSessionLocal

async def get_session() -> AsyncIterator[AsyncSession]:
    async with AsyncSessionLocal() as session:
        yield session

DBSession = Annotated[AsyncSession, Depends(get_session)]

이게 23장에서 본 Depends 패턴의 가장 강력한 활용. 라우트에서:

@router.get("/{todo_id}")
async def read_todo(todo_id: int, db: DBSession) -> TodoOut:
    ...

DBSession 한 줄이 들어가면 요청 단위로 세션이 만들어지고, 라우트가 끝나면 자동 닫힘. 트랜잭션 관리도 같은 곳에서 수행됩니다.

모델 정의 — 2.x 스타일 #

app/models/base.py

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

app/models/todo.py

from datetime import datetime
from sqlalchemy import String, Boolean, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column
from app.models.base import Base

class Todo(Base):
    __tablename__ = "todos"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    description: Mapped[str | None]
    done: Mapped[bool] = mapped_column(Boolean, default=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=func.now(),
    )

9장 typing 본격의 Mapped[T]가 정확히 여기에서 쓰입니다. 일반 타입 힌트처럼 보이지만 SQL 컬럼 정의가 같이 일어납니다.

읽는 법:

Mapped[int] — 컬럼의 파이썬 타입
mapped_column(...) — DB 측 추가 옵션 (제약, 기본값 등)
Mapped[str | None] — NULL 허용
server_default=func.now() — DB가 자동 채움

관계 — 1:N #

users와 todos 1:N

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(200), unique=True, index=True)
    hashed_password: Mapped[str]

    todos: Mapped[list["Todo"]] = relationship(back_populates="owner")

class Todo(Base):
    __tablename__ = "todos"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    owner: Mapped["User"] = relationship(back_populates="todos")

relationship(back_populates=...)가 양방향 연결을 만듭니다. user.todos, todo.owner 두 접근 모두 자연스럽게 동작합니다.

CRUD — 2.x 스타일 #

생성 #

create

from sqlalchemy.ext.asyncio import AsyncSession

async def create_todo(db: AsyncSession, data: TodoCreate, owner_id: int) -> Todo:
    todo = Todo(
        title=data.title,
        description=data.description,
        owner_id=owner_id,
    )
    db.add(todo)
    await db.commit()
    await db.refresh(todo)    # DB 가 채운 id, created_at 등을 다시 로드
    return todo

db.refresh(obj)가 DB가 자동 생성한 컬럼을 다시 읽어옵니다 (id, timestamp 등). 비동기에서 자주 빠뜨리는 부분입니다.

조회 #

select

from sqlalchemy import select

async def get_todo(db: AsyncSession, todo_id: int) -> Todo | None:
    result = await db.execute(
        select(Todo).where(Todo.id == todo_id)
    )
    return result.scalar_one_or_none()

async def list_todos(db: AsyncSession, skip: int, limit: int) -> list[Todo]:
    result = await db.execute(
        select(Todo).offset(skip).limit(limit).order_by(Todo.created_at.desc())
    )
    return list(result.scalars())

핵심 패턴:

select(Todo).where(...) — 쿼리 객체 (실행 안 됨)
db.execute(...) — 실제 실행
.scalar_one_or_none() — 0 또는 1 행, 없으면 None
.scalars() — 모든 행을 ORM 객체로

옛 1.x의 db.query(Todo).filter(...).first() 패턴은 2.x에서 deprecated입니다. 새 코드는 무조건 select + execute.

업데이트 #

update — 객체 수정

async def update_todo(db: AsyncSession, todo: Todo, data: TodoUpdate) -> Todo:
    update_dict = data.model_dump(exclude_unset=True)
    for field, value in update_dict.items():
        setattr(todo, field, value)
    await db.commit()
    await db.refresh(todo)
    return todo

또는 SQL UPDATE 직접:

update — SQL 직접

from sqlalchemy import update

async def mark_all_done(db: AsyncSession, owner_id: int) -> int:
    result = await db.execute(
        update(Todo)
        .where(Todo.owner_id == owner_id)
        .values(done=True)
    )
    await db.commit()
    return result.rowcount

ORM 객체 단위 업데이트는 변경 추적이 자동, SQL 직접은 대량 업데이트에 빠릅니다. 상황에 따라 선택.

삭제 #

delete

async def delete_todo(db: AsyncSession, todo: Todo) -> None:
    await db.delete(todo)
    await db.commit()

N+1 문제와 즉시 로딩 #

ORM의 가장 흔한 성능 함정. 사용자 100명 + 각자의 todos를 가져올 때:

🚫 N+1

users = (await db.execute(select(User))).scalars()
for user in users:
    todos = user.todos    # ← 매번 추가 쿼리

100명 + 100번 쿼리 = 101번 쿼리. selectinload 또는 **joinedload**로 한 번에:

✅ selectinload

from sqlalchemy.orm import selectinload

result = await db.execute(
    select(User).options(selectinload(User.todos))
)
users = result.scalars()
for user in users:
    print(user.todos)    # 추가 쿼리 없음

selectinload — 별도 쿼리로 가져옴 (1:N에 어울림)
joinedload — JOIN 한 번에 (1:1, N:1에 어울림)

21장 성능 — cProfile, py-spy, 메모리 프로파일링의 측정 흐름이 여기서도 — echo=True로 SQL 로그 켜고 N+1 패턴 보이면 잡으세요.

라우트와 결합 #

app/api/todos.py — DB 버전

from fastapi import APIRouter, HTTPException
from app.api.deps import DBSession
from app.schemas.todo import TodoCreate, TodoOut, TodoUpdate
from app.models.todo import Todo
from sqlalchemy import select

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

@router.post("/", response_model=TodoOut, status_code=201)
async def create_todo(payload: TodoCreate, db: DBSession) -> Todo:
    todo = Todo(**payload.model_dump(), owner_id=1)   # owner_id 는 26장에서 인증
    db.add(todo)
    await db.commit()
    await db.refresh(todo)
    return todo

@router.get("/{todo_id}", response_model=TodoOut)
async def read_todo(todo_id: int, db: DBSession) -> Todo:
    todo = await db.get(Todo, todo_id)
    if todo is None:
        raise HTTPException(404, "Todo not found")
    return todo

response_model=TodoOut + model_config = {"from_attributes": True} (23장에서 본 그것)가 SQLAlchemy 객체를 Pydantic으로 자동 변환합니다.

Alembic — 스키마 마이그레이션 #

DB 스키마는 시간이 지나며 바뀝니다. 컬럼 추가, 인덱스 변경, 테이블 새로 만들기 등. 그 변경을 버전 관리 하는 도구가 Alembic입니다.

초기화 #

alembic 초기화

uv run alembic init -t async alembic

-t async가 비동기 템플릿. 디렉터리 구조:

alembic/
├── env.py              # 마이그레이션 환경
├── script.py.mako       # 템플릿
└── versions/            # 마이그레이션 파일들

설정 #

alembic.ini — 일부만 수정

sqlalchemy.url = postgresql+asyncpg://user:pw@localhost/todo

또는 환경 변수에서 읽는 것이 안전합니다.

alembic/env.py — 핵심 부분

from app.core.config import settings
from app.models.base import Base
from app.models import todo, user   # 모든 모델 import

config = context.config
config.set_main_option("sqlalchemy.url", settings.database_url)

target_metadata = Base.metadata

target_metadata = Base.metadata 한 줄이 핵심 — Alembic이 모델 정의와 현재 DB 상태를 비교해서 마이그레이션을 자동 생성합니다.

첫 마이그레이션 #

자동 생성

uv run alembic revision --autogenerate -m "create todos table"

alembic/versions/에 새 파일이 생성됩니다.

versions/abc123_create_todos_table.py

def upgrade() -> None:
    op.create_table(
        "todos",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("title", sa.String(200)),
        ...
    )

def downgrade() -> None:
    op.drop_table("todos")

자동 생성된 코드를 반드시 검토 하세요. autogenerate가 모든 변화를 정확히 잡지는 않습니다. 특히 인덱스, 제약, 컬럼 타입 변경에 주의.

적용 #

upgrade

uv run alembic upgrade head    # 최신까지
uv run alembic downgrade -1    # 한 단계 되돌림
uv run alembic current         # 현재 버전
uv run alembic history         # 모든 마이그레이션

머지 충돌 해결 #

여러 개발자가 동시에 마이그레이션을 만들면 분기가 생깁니다. **alembic merge**로 합칩니다.

merge

uv run alembic merge -m "merge a and b" abc123 def456

자주 만나는 함정 #

1) `select` 결과를 `.all()`로 받지 말 것 #

🚫

result = await db.execute(select(Todo))
rows = result.all()    # 각 row 가 (Todo,) 튜플

✅

result = await db.execute(select(Todo))
todos = list(result.scalars())    # Todo 객체 리스트

.scalars()가 ORM 객체를 직접 풀어줍니다.

2) 비동기 세션 안에서 동기 lazy load 금지 #

expire_on_commit=False + selectinload / joinedload로 미리 로드. 그렇지 않으면 라우트가 응답을 만들 때 lazy load가 트리거되어 MissingGreenlet 에러가 납니다.

3) 트랜잭션 누락 #

commit을 안 부르면 변경이 반영되지 않습니다. 의존성에서 자동 commit 하는 패턴도 가능:

자동 commit 의존성

async def get_session() -> AsyncIterator[AsyncSession]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

다만 명시적 commit이 더 안전한 경우도 있습니다. 프로젝트 정책으로 정하세요.

연습문제 #

SQLite + aiosqlite로 todo-api 프로젝트에 app/db/session.py, app/models/{base,todo,user}.py를 작성하세요. Mapped[T] + mapped_column(...) 스타일로 컬럼 정의. SQL echo를 켜고 첫 select를 실행해 로그를 직접 확인합니다.
User ↔ Todo 1:N 관계를 정의하세요. 일부러 N+1 쿼리가 나는 코드를 작성한 뒤 selectinload(User.todos)로 고쳐 SQL 로그에서 쿼리 횟수 변화를 확인합니다.
Alembic을 초기화하고 (init -t async) 첫 마이그레이션을 자동 생성합니다 (revision --autogenerate). 생성된 파일을 검토한 뒤 upgrade head로 적용하고, downgrade -1로 되돌리는 과정까지 직접 확인합니다.

한 줄 요약: SQLAlchemy 2.x의 Mapped[T] + mapped_column(...)가 표준. 비동기는 create_async_engine + AsyncSession, expire_on_commit=False. Depends 의존성으로 요청 단위 세션. CRUD는 select(...).where(...) + db.execute(...), .scalars() / .scalar_one_or_none(). 관계는 relationship(back_populates=...), N+1은 selectinload / joinedload로 회피. Pydantic from_attributes=True로 ORM → 응답 자동. Alembic으로 스키마 마이그레이션 버전 관리.

다음 챕터 #

다음 26장 인증 — OAuth2 패스워드 플로우 + JWT에서는 본 챕터 위에 사용자 회원가입 / 로그인을 얹습니다. OAuth2 패스워드 플로우, JWT 발급, 비밀번호 해싱, 그리고 의존성으로 구성하는 current_user 까지입니다.