SQL(관계형) 데이터베이스

SQL(관계형) 데이터베이스 (SQL Databases)

API를 만들다 보면 데이터를 어딘가에 저장해야 하는 순간이 거의 반드시 와요. 이번 장에서는 SQLite를 써서 FastAPI와 SQL(관계형) 데이터베이스를 연결하는 방법을 배워요. SQLite는 파일 하나로 동작하고 Python이 기본 지원을 갖고 있어서, 예제를 그대로 복사해서 바로 돌려볼 수 있다는 장점이 있죠.

출처: 공식문서

나중에 실제 운영(production)에서는 PostgreSQL 같은 데이터베이스 서버를 쓰게 되겠지만, 개념은 똑같아요. 이번 장은 그 연결 구조를 이해하는 데 집중할게요.

팁

PostgreSQL과 FastAPI를 쓰는 공식 프로젝트 생성기를 제공하고 있어요. 프론트엔드와 여러 도구까지 포함되어 있으니 나중에 참고해도 좋아요.

https://github.com/fastapi/full-stack-fastapi-template

이 예제에서는 SQLModel을 사용해요. SQLModel은 SQLAlchemy와 Pydantic을 하나로 합친 라이브러리로, 모델 클래스 하나로 데이터베이스 테이블과 데이터 검증을 동시에 다룰 수 있게 해 줘요.

모델 하나로 앱 만들기

먼저 가장 단순한 버전부터 시작해요. 모델 하나(Hero)와 CRUD 동작들을 만들어 볼게요.

모델 만들기

SQLModel을 상속하고 table=True를 붙인 클래스가 하나의 테이블 모델이 돼요. str로 선언된 필드는 데이터베이스에서 TEXT(또는 데이터베이스에 따라 VARCHAR) 컬럼이 되고, int는 정수 컬럼이 되는 식이에요. 우린 타입만 선언하면 되죠.

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    age: int | None = Field(default=None, index=True)
    secret_name: str

index=True는 이 컬럼에 데이터베이스 인덱스를 만들어서 조회를 빠르게 해 주는 옵션이에요.

엔진(Engine) 만들기

create_engine()으로 데이터베이스에 연결하는 엔진을 만들어요. 여기서 check_same_thread=False를 넘기는 이유가 약간 미묘한데, 천천히 설명할게요.

FastAPI에서는 하나의 요청이 여러 개의 스레드를 거칠 수 있어요(예: 의존성 처리 중). SQLite는 기본적으로 한 스레드에서만 접근하도록 되어 있어서, 이 옵션을 켜서 여러 스레드에서 같은 SQLite 데이터베이스를 쓸 수 있게 풀어 주는 거예요. 다만 코드 구조상 **요청마다 단일 Session**을 쓰도록 만들 거라서, 실제로는 같은 세션이 여러 스레드를 오가는 일이 없게 돼요. 지금은 "옵션을 켰지만 구조적으로 안전하게 쓴다"는 정도로 이해하면 돼요.

sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

SQLModel.metadata.create_all(engine)는 아직 없는 테이블들을 만들어 주는 함수예요.

세션 의존성 만들기

**Session**은 객체를 메모리에 저장하고, 데이터에 필요한 변경을 추적한 뒤, engine을 통해 데이터베이스와 통신하는 역할을 해요.

요청마다 새 Session을 제공하는 FastAPI 의존성을 yield로 만들어 볼게요. yield를 쓰면 요청이 끝나고 with 블록이 닫히면서 세션이 정리되는데, 이렇게 해서 요청당 단일 세션을 보장해요.

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


SessionDep = Annotated[Session, Depends(get_session)]

SessionDep = Annotated[Session, Depends(get_session)]처럼 하나로 묶어 두면, 이후에 모든 _경로 동작_에서 session: SessionDep이라고만 써도 깔끔하게 재사용할 수 있어요.

시작 시 테이블 만들기

애플리케이션이 시작될 때 테이블을 생성하도록 on_event("startup")을 걸어요. 앱이 뜰 때 create_db_and_tables()가 실행되는 구조예요.

app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()

CRUD 동작 만들기

이제 모델과 세션, 테이블 준비가 끝났으니 실제 동작들을 만들면 돼요. session에 SessionDep을 주입해서 쓰는 게 전부예요.

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    age: int | None = Field(default=None, index=True)
    secret_name: str


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


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


SessionDep = Annotated[Session, Depends(get_session)]

app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/")
def create_hero(hero: Hero, session: SessionDep) -> Hero:
    session.add(hero)
    session.commit()
    session.refresh(hero)
    return hero


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


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


@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"ok": True}

동작의 흐름을 짚어 보면, session.add()로 새 객체를 넣고 commit()으로 저장하며, session.exec(select(Hero)...)로 조회해요. 한 가지 주의할 점은 create_hero가 받는 모델이 Hero 테이블 모델이라는 거예요. 즉 클라이언트가 secret_name까지 포함해서 데이터를 통째로 보내야 해요. 뒤에서 이걸 더 세련되게 나눠 볼게요.

여러 모델로 앱 업그레이드하기

지금 구조의 문제는 클라이언트가 secret_name 같은 민감한 필드까지 직접 보내야 한다는 점이에요. 그래서 실제로는 데이터를 여러 모델로 나눠서 다루는 게 일반적이에요. SQLModel에서는 요청·응답·DB 각각에 맞는 모델을 나눠 만들어요.

여러 모델 만들기

SQLModel에서 table=True를 갖는 클래스는 테이블 모델(DB에 저장되는 그대로), table=True가 없는 클래스는 데이터 모델(사실상 Pydantic 모델)이에요. 이 구분이 핵심이에요.

이 예제에서는 이렇게 나눠요.

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select


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


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


class HeroPublic(HeroBase):
    id: int


class HeroCreate(HeroBase):
    secret_name: str


class HeroUpdate(HeroBase):
    name: str | None = None
    age: int | None = None
    secret_name: str | None = None

하나씩 역할을 보면 이해가 빨라져요.

HeroBase: name과 age처럼 모든 모델이 공통으로 갖는 필드를 담은 뼈대예요.
Hero : HeroBase를 상속하면서 table=True를 붙인 테이블 모델이에요. 그래서 HeroBase의 필드에 더해 id, secret_name까지 모두 갖고, 실제 DB에 저장되는 대상이죠.
HeroCreate: 클라이언트가 새 영웅을 만들 때 보내는 데이터 모델이에요. 검증만 담당하고 secret_name을 요구해요. id는 필요 없죠.
HeroPublic: 응답으로 클라이언트에게 보여줄 데이터 모델이에요. id는 보여주되 secret_name은 숨겨요.
HeroUpdate: 수정 시 일부 필드만 받을 수 있게 모든 필드를 선택적으로 만든 모델이에요.

이렇게 나누면 "클라이언트는 HeroCreate를 보내고, 응답은 HeroPublic으로 받고, DB에는 Hero로 저장"이라는 흐름이 깔끔하게 나와요.

`HeroCreate`로 만들고 `HeroPublic`을 돌려주기

이제 응답 모델을 HeroPublic으로 지정해서, 민감한 secret_name은 응답에서 빠지도록 만들어요. 응답으로 response_model=HeroPublic을 선언하면 FastAPI가 HeroPublic으로 데이터를 검증하고 직렬화해요.

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select


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


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


class HeroPublic(HeroBase):
    id: int


class HeroCreate(HeroBase):
    secret_name: str


class HeroUpdate(HeroBase):
    name: str | None = None
    age: int | None = None
    secret_name: str | None = None


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


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


SessionDep = Annotated[Session, Depends(get_session)]
app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@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,
):
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes


@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


@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):
    hero_db = session.get(Hero, hero_id)
    if not hero_db:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero_data = hero.model_dump(exclude_unset=True)
    hero_db.sqlmodel_update(hero_data)
    session.add(hero_db)
    session.commit()
    session.refresh(hero_db)
    return hero_db


@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"ok": True}

새로 보이는 부분만 짚을게요.

Hero.model_validate(hero) — 받은 HeroCreate 데이터를 Hero 테이블 모델로 변환해요.
update_hero에서는 hero.model_dump(exclude_unset=True)로 클라이언트가 실제로 보낸 필드만 뽑아서, 안 보낸 필드는 건드리지 않아요. 그래서 PATCH(부분 수정)가 의도대로 동작하죠.
모든 응답이 HeroPublic이므로, secret_name은 절대 응답에 나타나지 않아요.

요약

이번 장을 통해 FastAPI + SQLModel의 핵심 흐름을 잡았어요. 요청·응답·DB용 모델을 나눠 만들고, response_model로 노출 범위를 통제하며, yield 의존성으로 요청당 단일 세션을 유지하는 구조요. 훨씬 더 깊은 내용은 SQLModel 공식 문서에 있는 FastAPI와 함께 쓰는 SQLModel 튜토리얼에서 다뤄져 있어요.

더 알아보기 (Learn more)

공식문서: SQL Databases
SQLModel 공식 문서: SQLModel Tutorial - FastAPI
Full Stack FastAPI 템플릿: fastapi/full-stack-fastapi-template

SQL(관계형) 데이터베이스