1. 프로젝트에 설치

앞서 배운 내용을 바탕으로 FastAPI 프로젝트에 SQLAlchemy를 적용해보겠습니다. 3장에서 만든 CRUD를 그대로 데이터베이스 버전으로 바꾸는 것이 목표입니다.

앞 실습의 서버가 실행 중이면 Ctrl + C로 멈추고, 가상환경이 켜져 있다면 deactivate로 빠져나옵니다. 실습 폴더들을 모아둔 상위 폴더에서 터미널에 다음 명령을 실행하세요.

mkdir 04_2_db
cd 04_2_db
python -m venv venv
.\venv\Scripts\Activate.ps1
pip install "fastapi[standard]" sqlalchemy

macOS/Linux에서는 python -m venv venv 대신 python3 -m venv venv를, 활성화 명령 대신 source ./venv/bin/activate를 사용합니다. 이후 명령은 가상환경이 활성화된 상태에서 실행합니다.

SQLite는 Python에 기본으로 포함되어 있으므로 별도로 설치할 필요가 없습니다.

2. 데이터베이스 설정

2.1 데이터베이스 URL 설정

DB를 연결할 수 있는 코드를 작성합니다. 보통은 database.py 파일을 생성하고 내용을 작성하지만 우리는 main.py 파일에 작성하겠습니다. 파일을 나누는 것은 6장에서 다룹니다. 전체 코드는 마지막에 제공하니 일일이 붙여넣기를 하지 않아도 됩니다. 또한 모든 코드의 동작 원리를 이해하기 보다는 전체 코드의 맥락을 이해하고, 동작을 확인한 다음, 필요한 부분을 수정하거나 추가하는 것을 목표로 하세요.

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

# Database setup
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},
)
SessionLocal = sessionmaker(bind=engine)


class Base(DeclarativeBase):
    pass

여기서 SQLALCHEMY_DATABASE_URL은 SQLite 데이터베이스 파일의 경로를 지정합니다. sqlite:///./sql_app.db는 현재 디렉토리에 sql_app.db 파일을 생성합니다. 앞에 sqlite:///까지가 프리픽스 입니다. sqlite:///./sql_app.db와 sqlite:///sql_app.db는 같습니다. ./는 현재 폴더라는 의미입니다. 따라서 현재 디렉토리에 sql_app.db 파일이 생성됩니다.

engine은 연결 설정이라고 생각하시면 됩니다. create_engine 함수를 사용하여 데이터베이스와 연결을 설정합니다. 이 함수는 멀티 쓰레드 사용 등 다양한 옵션을 지원합니다. 여기서 connect_args={"check_same_thread": False}가 중요합니다.

sqlite3.ProgrammingError: SQLite objects created in a thread can only be used in that same thread.

SessionLocal은 데이터베이스 세션을 생성하는데 사용됩니다. 세션은 데이터베이스 연결을 나타내며, SessionLocal()을 호출할 때마다 새 세션이 하나씩 만들어집니다. 각 코드는 독립적인 세션을 사용할 수 있습니다.

Base는 모델을 정의하는데 사용됩니다. 모델은 데이터베이스 테이블을 정의하는데 사용됩니다.

2.2 데이터베이스 세션 관리

main.py 세션을 관리할 수 있는 코드 입니다. 여기서 get_db는 FastAPI의 함수가 호출할 수 있도록(의존성 주입) 코드를 작성할 것입니다. 이렇게 호출된 함수는 데이터베이스 세션을 생성하고, 세션을 닫아주는 역할을 합니다. 세션을 닫아주지 않으면 데이터베이스 연결이 계속 유지되어 연결이 쌓이고, 결국 서버가 응답하지 못하게 됩니다.

from collections.abc import Generator

from sqlalchemy.orm import Session


# Dependency
def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

여기서 yield가 쓰인 이유가 있습니다. 일반적인 return이면 값을 돌려주고 함수가 끝나버려서 뒷정리를 할 기회가 없습니다. yield를 쓰면 값을 넘겨준 뒤 잠시 멈춰 있다가, 요청 처리가 끝나면 다시 깨어나 finally 블록을 실행합니다. 예시를 통해 알아보겠습니다.

@app.get("/users")
def read_users(db: SessionDep):
    # 1. get_db()가 호출되어 새 세션 생성
    # 이제 db로 이 세션에 접근할 수 있음
    # 2. yield를 통해 이 세션이 여기로 전달됨
    users = db.scalars(select(User)).all()
    return users
    # 3. 함수 실행이 끝나면 finally 블록이 실행되어 세션이 종료됨

조금 더 어려운 말로 @app.get("/users")를 라우트 핸들러라고 하는데 이 라우트 핸들러의 작업이 완료되면 finally 블록이 실행되어 세션을 종료하는 것입니다. SessionDep이 무엇인지는 5절에서 설명합니다.

3. 모델 정의

모델은 데이터베이스 테이블을 정의하는데 사용됩니다. 보통은 models.py 파일을 생성하고 정의합니다. 우리는 main.py 파일에 작성하겠습니다.

from sqlalchemy.orm import Mapped, mapped_column


# Model
class Item(Base):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    name: Mapped[str] = mapped_column(index=True)
    description: Mapped[str | None]
    price: Mapped[float]

여기서 Item은 데이터베이스 테이블을 정의하는데 사용됩니다. __tablename__은 테이블 이름을 지정합니다. id, name, description, price는 각각 컬럼을 정의합니다. Mapped[int], Mapped[str], Mapped[float]은 데이터 타입을 지정하고, | None이 붙으면 비어 있어도 되는 컬럼입니다. primary_key=True는 자동으로 증가되는 숫자값인 기본키를 지정합니다. index=True는 인덱스를 지정합니다. 인덱스는 그 컬럼으로 검색할 일이 많을 때 붙입니다. 책 뒤의 색인처럼, 전체를 훑지 않고 빠르게 찾을 수 있게 해줍니다. 대신 데이터를 추가할 때 색인도 함께 갱신해야 하므로 모든 컬럼에 붙이는 것은 좋지 않습니다.

# Create tables
Base.metadata.create_all(bind=engine)

모델을 정의했다면 데이터베이스를 생성해야 합니다. metadata.create_all를 하게 되면 실제 데이터베이스에 테이블이 생성됩니다. 이 코드는 "정의한 클래스에 해당하는 테이블이 없으면 만들어라"라는 뜻입니다. 이미 있으면 아무것도 하지 않습니다.

4. Pydantic 스키마 정의

스키마라는 단어는 데이터베이스에서 사용되는 용어입니다. 데이터베이스의 테이블 구조를 정의하는데 사용됩니다. Pydantic 스키마는 데이터를 검증하고 파싱하는데 사용됩니다. 우리는 앞서 3-1 챕터에서 BaseModel을 사용하여 Pydantic으로된 스키마를 어떻게 사용하는지 배웠습니다. 보통은 schemas.py 파일을 생성하고 정의하지만 main.py 파일에 작성하겠습니다.

주의하셔야 할 것은 Pydantic으로 정의된 스키마는 ORM 모델과는 다릅니다. ORM 모델은 데이터베이스 테이블을 정의하는데 사용되고, Pydantic 스키마는 데이터를 검증하고 파싱하는데 사용됩니다. ORM 모델과 Pydantic 스키마는 서로 다른 역할을 가지고 있습니다. 따라서 이 두개의 모델은 변환을 해주어야 합니다.

from pydantic import BaseModel, ConfigDict, Field


# Pydantic schema
class ItemCreate(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    description: str | None = None
    price: float = Field(ge=0)


class ItemUpdate(BaseModel):
    name: str | None = Field(default=None, min_length=1, max_length=50)
    description: str | None = None
    price: float | None = Field(default=None, ge=0)


class ItemPublic(ItemCreate):
    model_config = ConfigDict(from_attributes=True)

    id: int

model_config = ConfigDict(from_attributes=True)가 새로 나왔습니다. 이것이 없으면 SQLAlchemy 객체를 Pydantic 모델로 바꿀 수 없습니다. Pydantic은 기본적으로 딕셔너리를 받아들이는데, SQLAlchemy 객체는 딕셔너리가 아니라 속성을 가진 객체이기 때문입니다. 이 설정은 "속성에서 값을 읽어도 된다"고 알려주는 역할을 합니다.

# 예전 방식
class ItemPublic(BaseModel):
    id: int

    class Config:
        orm_mode = True  # v1 초기
        # 또는
        from_attributes = True  # v1 후기

5. CRUD 작업 구현

실제 CRUD 작업을 구현하기 전에 처음 보는 코드를 먼저 설명하겠습니다. 세션을 쓰는 엔드포인트마다 Depends(get_db)를 적으면 반복이 심합니다. 타입 별칭으로 한 번만 정의해두면 훨씬 짧아집니다.

from typing import Annotated

from fastapi import Depends

SessionDep = Annotated[Session, Depends(get_db)]

이제 엔드포인트에서 db: SessionDep이라고만 적으면 됩니다. 이 부분을 FastAPI의 의존성 주입이라고 합니다. Annotated[Session, Depends(get_db)]는 두 부분으로 나눌 수 있습니다.

1. Session: 데이터베이스 세션 타입 명시
2. Depends(get_db): 의존성 주입 선언

여기서 Depends는 FastAPI에게 "이 함수가 실행되기 전에 먼저 get_db() 함수를 실행해서 그 결과를 db 매개변수로 주입해줘"라고 지시합니다. get_db() 함수를 다시 한 번 살펴보도록 하겠습니다. 여기서 SessionLocal()로 데이터베이스 연결을 하고 이 연결을 yield로 반환합니다. 이렇게 하면 FastAPI가 get_db() 함수를 실행하고 반환된 데이터베이스 세션을 db 매개변수로 주입해줍니다.

# Dependency
def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

그리고 엔드포인트 함수가 종료될 때 위 함수의 finally 블록이 실행되어 데이터베이스 세션이 종료됩니다. 이렇게 하면 데이터베이스 연결이 계속 유지되어 메모리 누수가 발생하지 않습니다.

이렇게 사용하게 되면 여러 엔드포인트에서 동일한 의존성을 쉽게 재사용할 수 있습니다. 여기서 의존성이라는 단어는 다른 함수나 클래스에서 사용하는 함수나 클래스를 의미합니다. 이렇게 하면 코드를 재사용하고, 중복을 줄일 수 있습니다. 또 테스트할 때 진짜 데이터베이스 대신 테스트용 데이터베이스로 바꿔 끼울 수 있습니다. 이 부분은 7장에서 실제로 해봅니다.

# 예전 방식
def read_items(db: Session = Depends(get_db)):
    ...

이제 실제 CRUD 작업을 구현합니다.

from fastapi import FastAPI, HTTPException, status
from sqlalchemy import select

app = FastAPI(title="물품 관리 API (DB 버전)")


# CRUD operations
@app.post("/items", status_code=status.HTTP_201_CREATED)
def create_item(item_data: ItemCreate, db: SessionDep) -> ItemPublic:
    db_item = Item(**item_data.model_dump())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item


@app.get("/items")
def read_items(db: SessionDep) -> list[ItemPublic]:
    return list(db.scalars(select(Item)).all())


@app.get("/items/{item_id}")
def read_item(item_id: int, db: SessionDep) -> ItemPublic:
    db_item = db.get(Item, item_id)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item


@app.patch("/items/{item_id}")
def update_item(item_id: int, item_data: ItemUpdate, db: SessionDep) -> ItemPublic:
    db_item = db.get(Item, item_id)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")

    update_data = item_data.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        setattr(db_item, key, value)

    db.commit()
    db.refresh(db_item)
    return db_item


@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(item_id: int, db: SessionDep) -> None:
    db_item = db.get(Item, item_id)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    db.delete(db_item)
    db.commit()

코드에서 눈여겨볼 곳이 몇 군데 있습니다.

db.add() 다음에 db.commit(), 그다음에 db.refresh()입니다.

refresh를 빼면 db_item.id를 읽을 때 문제가 생길 수 있습니다. id는 데이터베이스가 정해주는 값이라 파이썬 객체에는 아직 없기 때문입니다.

조회에는 db.get()을 썼습니다. 기본키로 하나를 찾을 때는 select보다 db.get(Item, item_id)가 간결합니다. 없으면 None을 반환합니다.

수정에는 setattr을 썼습니다. 3장의 메모리 버전에서는 model_copy(update=...)로 새 객체를 만들었습니다. 데이터베이스 버전에서는 새 객체를 만들면 안 됩니다. 세션이 추적하고 있는 그 객체를 직접 고쳐야 변경사항이 저장되기 때문입니다. exclude_unset=True로 걸러낸 필드만 하나씩 덮어씁니다.

6. 전체 코드

main.py 파일에 붙여넣고 실행해보세요.

from collections.abc import Generator
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, status
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import create_engine, select
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    Session,
    mapped_column,
    sessionmaker,
)

# ------------------------------------------------------------------
# 데이터베이스 설정
# ------------------------------------------------------------------
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},
)
SessionLocal = sessionmaker(bind=engine)


class Base(DeclarativeBase):
    pass


# ------------------------------------------------------------------
# 데이터베이스 모델
# ------------------------------------------------------------------
class Item(Base):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    name: Mapped[str] = mapped_column(index=True)
    description: Mapped[str | None]
    price: Mapped[float]


Base.metadata.create_all(bind=engine)


# ------------------------------------------------------------------
# Pydantic 스키마
# ------------------------------------------------------------------
class ItemCreate(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    description: str | None = None
    price: float = Field(ge=0)


class ItemUpdate(BaseModel):
    name: str | None = Field(default=None, min_length=1, max_length=50)
    description: str | None = None
    price: float | None = Field(default=None, ge=0)


class ItemPublic(ItemCreate):
    model_config = ConfigDict(from_attributes=True)

    id: int


# ------------------------------------------------------------------
# 의존성
# ------------------------------------------------------------------
def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


SessionDep = Annotated[Session, Depends(get_db)]


# ------------------------------------------------------------------
# 엔드포인트
# ------------------------------------------------------------------
app = FastAPI(title="물품 관리 API (DB 버전)", version="1.0.0")


@app.post("/items", status_code=status.HTTP_201_CREATED, tags=["물품"])
def create_item(item_data: ItemCreate, db: SessionDep) -> ItemPublic:
    """새 물품을 데이터베이스에 저장합니다."""
    db_item = Item(**item_data.model_dump())
    db.add(db_item)  # 세션에 올려둡니다(아직 저장되지 않았습니다)
    db.commit()  # 실제로 저장합니다
    db.refresh(db_item)  # DB가 채워준 id를 가져옵니다
    return db_item


@app.get("/items", tags=["물품"])
def read_items(db: SessionDep) -> list[ItemPublic]:
    """모든 물품을 반환합니다."""
    return list(db.scalars(select(Item)).all())


@app.get("/items/{item_id}", tags=["물품"])
def read_item(item_id: int, db: SessionDep) -> ItemPublic:
    """ID로 물품 하나를 조회합니다."""
    db_item = db.get(Item, item_id)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item


@app.patch("/items/{item_id}", tags=["물품"])
def update_item(item_id: int, item_data: ItemUpdate, db: SessionDep) -> ItemPublic:
    """보낸 필드만 수정합니다."""
    db_item = db.get(Item, item_id)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")

    for key, value in item_data.model_dump(exclude_unset=True).items():
        setattr(db_item, key, value)

    db.commit()
    db.refresh(db_item)
    return db_item


@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT, tags=["물품"])
def delete_item(item_id: int, db: SessionDep) -> None:
    """물품을 삭제합니다."""
    db_item = db.get(Item, item_id)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    db.delete(db_item)
    db.commit()

7. 애플리케이션 실행

이제 애플리케이션을 실행할 수 있습니다.

fastapi dev

이제 FastAPI 애플리케이션이 SQLite 데이터베이스와 연동되어 실행됩니다. 아래 URL로 접속하여 API를 테스트해보세요.

• http://127.0.0.1:8000/docs

.http 파일로 테스트한다면 api.http 파일을 만들어 3장과 같은 순서로 요청을 보내보세요.

@baseUrl = http://127.0.0.1:8000

### 1. 물품 등록
POST {{baseUrl}}/items
Content-Type: application/json

{
    "name": "item1",
    "description": "item1 description",
    "price": 100
}

### 2. 목록 조회
GET {{baseUrl}}/items

### 3. 상세 조회
GET {{baseUrl}}/items/1

### 4. 가격만 수정
PATCH {{baseUrl}}/items/1
Content-Type: application/json

{
    "price": 999
}

### 5. 삭제
DELETE {{baseUrl}}/items/1

3장과 결정적으로 다른 점을 확인해보겠습니다. 물품을 몇 개 등록한 다음, 터미널에서 Ctrl + C로 서버를 완전히 끄고 다시 실행해보세요.

fastapi dev

GET /items를 다시 보내면 데이터가 그대로 남아 있습니다. 3장에서는 사라졌던 데이터입니다. 이것이 데이터베이스를 쓰는 이유입니다.

8. 데이터 확인

데이터를 확인하는 방법으로는 SQLite 데이터베이스 파일을 직접 열어서 확인하는 방법이 있습니다. 폴더에 sql_app.db 파일이 생성되었을 것입니다. 더블클릭을 하면 바로 열리는 것은 아니기 때문에 SQLite Viewer 익스텐션을 설치하거나 DB Browser for SQLite 등의 프로그램을 사용하면 쉽게 확인할 수 있습니다.

.gitignore 파일에 아래 내용을 추가해두는 것을 권합니다.

*.db
venv/
__pycache__/

연습문제

1. Item 모델에 created_at 필드를 추가하고, 물품 생성 시 자동으로 현재 시간이 저장되도록 구현해보세요. 힌트로 mapped_column(insert_default=func.now())를 사용할 수 있습니다.

2. 이름으로 검색하는 기능을 추가해보세요. GET /items?q=키보드 형태이며, 힌트로 select(Item).where(Item.name.contains(q))를 사용합니다.

3. 물품 목록에 페이지네이션을 추가해보세요. 2장에서 배운 skip과 limit을 쿼리 매개변수로 받고, select(Item).offset(skip).limit(limit)을 사용합니다.

4. 새로운 모델(예: User)을 추가하고, 이에 대한 CRUD 작업과 API 엔드포인트를 구현해보세요.

5. SQLite 데이터베이스 파일의 경로를 코드에 직접 적지 않고 환경 변수에서 읽어오도록 코드를 수정해보세요. 제대로 하는 방법은 7장에서 다루지만, 먼저 시도해보시면 좋습니다.