모던 파이썬 실전 #3 DB 연동 — SQLAlchemy 2.x + Alembic
#2의 인메모리 dict를 이번 글에서 진짜 DB로 교체합니다. SQLAlchemy 2.x의 새 스타일, async 세션, FastAPI 의존성 주입과의 자연스러운 결합, 그리고 스키마 변경을 안전하게 굴리는 Alembic 마이그레이션까지 다룹니다.
왜 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 별도를 권장합니다.
설치 #
uv add sqlalchemy "sqlalchemy[asyncio]" asyncpg
uv add --dev alembic- sqlalchemy — ORM 본체
- asyncpg — PostgreSQL 비동기 드라이버 (async 사용 시)
- alembic — 마이그레이션 도구
SQLite로 시작하려면 aiosqlite (PostgreSQL 대신).
uv add aiosqliteEngine과 세션 #
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 함수 밖에서는 위험합니다.
의존성으로 세션 주입 #
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)]이게 #2에서 본 Depends 패턴의 가장 강력한 활용. 라우트에서:
@router.get("/{todo_id}")
async def read_todo(todo_id: int, db: DBSession) -> TodoOut:
...DBSession 한 줄이 들어가면 요청 단위로 세션이 만들어지고, 라우트가 끝나면 자동 닫힘. 트랜잭션 관리도 같은 곳에서 수행됩니다.
모델 정의 — 2.x 스타일 #
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
passfrom 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(),
)중급 #2의 Mapped[T]가 정확히 여기에서 쓰입니다. 일반 타입 힌트처럼 보이지만 SQL 컬럼 정의가 같이 일어납니다.
읽는 법:
Mapped[int]— 컬럼의 파이썬 타입mapped_column(...)— DB 측 추가 옵션 (제약, 기본값 등)Mapped[str | None]— NULL 허용server_default=func.now()— DB가 자동 채움
관계 — 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 스타일 #
생성 #
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 tododb.refresh(obj)가 DB가 자동 생성한 컬럼을 다시 읽어옵니다 (id, timestamp 등). 비동기에서 자주 빠뜨리는 부분입니다.
조회 #
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.
업데이트 #
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 직접:
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.rowcountORM 객체 단위 업데이트는 변경 추적이 자동, SQL 직접은 대량 업데이트에 빠릅니다. 상황에 따라 선택.
삭제 #
async def delete_todo(db: AsyncSession, todo: Todo) -> None:
await db.delete(todo)
await db.commit()N+1 문제와 즉시 로딩 #
ORM의 가장 흔한 성능 함정. 사용자 100명 + 각자의 todos를 가져올 때:
users = (await db.execute(select(User))).scalars()
for user in users:
todos = user.todos # ← 매번 추가 쿼리100명 + 100번 쿼리 = 101번 쿼리. selectinload 또는 **joinedload**로 한 번에:
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에 어울림)
고급 #7의 측정 흐름이 여기서도 — echo=True로 SQL 로그 켜고 N+1 패턴 보이면 잡으세요.
라우트와 결합 #
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는 #4에서 인증
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 todoresponse_model=TodoOut + model_config = {"from_attributes": True} (#2에서 본 그것)가 SQLAlchemy 객체를 Pydantic으로 자동 변환합니다.
Alembic — 스키마 마이그레이션 #
DB 스키마는 시간이 지나며 바뀝니다. 컬럼 추가, 인덱스 변경, 테이블 새로 만들기 등. 그 변경을 버전 관리하는 도구가 Alembic입니다.
초기화 #
uv run alembic init -t async alembic-t async가 비동기 템플릿. 디렉터리 구조:
alembic/
├── env.py # 마이그레이션 환경
├── script.py.mako # 템플릿
└── versions/ # 마이그레이션 파일들설정 #
sqlalchemy.url = postgresql+asyncpg://user:pw@localhost/todo또는 환경 변수에서 읽는 것이 안전합니다.
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.metadatatarget_metadata = Base.metadata 한 줄이 핵심 — Alembic이 모델 정의와 현재 DB 상태를 비교해서 마이그레이션을 자동 생성합니다.
첫 마이그레이션 #
uv run alembic revision --autogenerate -m "create todos table"alembic/versions/에 새 파일이 생성됩니다.
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가 모든 변화를 정확히 잡지는 않습니다. 특히 인덱스, 제약, 컬럼 타입 변경에 주의.
적용 #
uv run alembic upgrade head # 최신까지
uv run alembic downgrade -1 # 한 단계 되돌림
uv run alembic current # 현재 버전
uv run alembic history # 모든 마이그레이션머지 충돌 해결 #
여러 개발자가 동시에 마이그레이션을 만들면 분기가 생깁니다. **alembic 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 하는 패턴도 가능:
async def get_session() -> AsyncIterator[AsyncSession]:
async with AsyncSessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise다만 명시적 commit이 더 안전한 경우도 있습니다. 프로젝트 정책으로 정하세요.
정리 #
이번 글에서 잡은 것:
- SQLAlchemy 2.x —
Mapped[T]+mapped_column(...)의 새 스타일 - 비동기 —
create_async_engine,AsyncSession,await db.execute(...) - FastAPI 의존성 주입 —
get_session으로 요청 단위 세션 - CRUD —
select(...).where(...),db.add,db.commit,db.refresh,db.delete - 관계 —
relationship+back_populates,selectinload/joinedload로 N+1 회피 - Pydantic
from_attributes=True로 ORM → 응답 자동 - Alembic —
init -t async,revision --autogenerate,upgrade head - 함정:
.scalars(), expire_on_commit, lazy load, 명시적 commit
다음 글(#4 인증 — OAuth2 + JWT)에서는 이 위에 사용자 회원가입/로그인을 얹습니다. OAuth2 패스워드 플로우, JWT 발급, 비밀번호 해싱, 그리고 의존성으로 풀어내는 current_user까지 다룹니다.