데이터와 상태

FastAPI는 특정 데이터베이스나 ORM에 묶여 있지 않습니다. 이 장에서는 공식 문서가 예제로 쓰는 SQLModel로 데이터베이스를 연결하고, 의존성으로 세션을 관리하는 방법, pydantic-settings로 설정을 읽는 방법, 응답 뒤에 작업을 실행하는 백그라운드 작업을 다룹니다.

SQLModel과 SQLAlchemy

SQLModel은 FastAPI 제작자가 만든 라이브러리로, SQLAlchemy 위에 Pydantic을 결합해 하나의 클래스로 테이블과 API 모델을 함께 정의합니다. SQLAlchemy를 직접 쓰는 것도 흔한 선택이며, 세션을 의존성으로 넘기는 방식은 같습니다.

pip install sqlmodel

모델과 엔진 정의

table=True인 클래스는 데이터베이스 테이블이 되고, 그렇지 않은 클래스는 입력 · 출력용 데이터 모델로만 쓰입니다. 공통 필드를 기반 클래스에 두고 상속하면 중복을 줄일 수 있습니다.

# app/db.py
from typing import Annotated

from fastapi import Depends
from sqlmodel import Field, Session, SQLModel, create_engine


class HeroBase(SQLModel):
    name: str = Field(index=True)
    age: int | None = None


class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    secret_name: str


class HeroCreate(HeroBase):
    secret_name: str


class HeroPublic(HeroBase):
    id: int


engine = create_engine(
    "sqlite:///database.db",
    connect_args={"check_same_thread": False},  # SQLite에서만 필요
)


def get_session():
    with Session(engine) as session:
        yield session


SessionDep = Annotated[Session, Depends(get_session)]

get_session은 yield를 쓰는 의존성입니다. 요청마다 세션을 하나 열어 경로 작업 함수에 넘기고, 응답이 끝나면 with 블록이 세션을 닫습니다. 테이블은 앱이 시작할 때 lifespan에서 SQLModel.metadata.create_all(engine)으로 만들 수 있으며, 운영 환경에서는 Alembic 같은 마이그레이션 도구를 쓰는 것이 좋습니다.

CRUD 엔드포인트

세션 의존성을 받아 추가와 목록 조회, 단건 조회를 구현합니다. 응답 모델을 HeroPublic으로 지정했으므로 secret_name은 응답에 나가지 않습니다.

from typing import Annotated

from fastapi import FastAPI, HTTPException, Query
from sqlmodel import select

from app.db import Hero, HeroCreate, HeroPublic, SessionDep

app = FastAPI()


@app.post("/heroes", response_model=HeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero


@app.get("/heroes", response_model=list[HeroPublic])
def read_heroes(
    session: SessionDep,
    offset: int = 0,
    limit: Annotated[int, Query(le=100)] = 100,
):
    return session.exec(select(Hero).offset(offset).limit(limit)).all()


@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

이 예제는 동기 세션을 쓰므로 경로 작업 함수를 def로 선언했습니다. 비동기 드라이버(asyncpg 등)와 SQLAlchemy의 AsyncSession을 쓴다면 async def와 await를 사용합니다.

pydantic-settings로 설정 읽기

데이터베이스 주소나 비밀 키는 환경 변수로 넘기고, pydantic-settings의 BaseSettings로 타입을 갖춘 설정 객체를 만듭니다. 환경 변수 이름은 필드 이름과 대소문자 구분 없이 대응됩니다.

# app/config.py
from functools import lru_cache

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    app_name: str = "Hello FastAPI"
    database_url: str = "sqlite:///database.db"
    secret_key: str

    model_config = SettingsConfigDict(env_file=".env")


@lru_cache
def get_settings() -> Settings:
    return Settings()

pip install pydantic-settings로 설치합니다. secret_key처럼 기본값이 없는 필드가 환경 변수에도 없으면 시작할 때 검증 오류가 나므로, 설정 누락을 일찍 알아챌 수 있습니다. @lru_cache 덕분에 .env 파일은 한 번만 읽히며, 경로에서는 Annotated[Settings, Depends(get_settings)]로 받아 씁니다. 의존성으로 받으면 테스트에서 다른 설정으로 바꿔 끼우기도 쉽습니다.

백그라운드 작업

메일 발송이나 로그 기록처럼 응답을 기다리게 할 필요가 없는 일은 BackgroundTasks로 응답을 보낸 뒤에 실행합니다.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message + "\n")


@app.post("/send-notification/{email}")
def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, f"notification sent to {email}")
    return {"message": "Notification queued"}

백그라운드 작업은 같은 프로세스 안에서 실행되므로 가벼운 일에 알맞습니다. 오래 걸리거나 재시도가 필요한 작업, 서버가 재시작되어도 잃으면 안 되는 작업은 Celery, ARQ 같은 별도 작업 큐를 쓰는 것이 좋습니다.

정리

• 데이터베이스 세션은 yield 의존성으로 요청마다 열고 닫습니다.
• SQLModel로 테이블 모델과 입력 · 출력 모델을 나누어 정의합니다.
• 설정은 pydantic-settings의 BaseSettings와 .env로 읽고, 의존성으로 주입합니다.
• 응답 뒤의 가벼운 작업은 BackgroundTasks, 무거운 작업은 별도 작업 큐로 처리합니다.