FastAPI Dependency로 세션 다루기

FastAPI Dependency로 세션 다루기 (Session with FastAPI Dependency)

뭔가를 더 추가하기 전에, 각 요청에 대해 세션을 얻는 방식을 좀 바꿔서 나중에 우리의 삶을 조금 단순하게 만들어 볼게요.

출처: 공식문서

현재의 세션들

지금까지 우리는 각 path operation 안의 with 블록에서 세션을 만들고 있었어요.

# Code above omitted 👆

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

# Code below omitted 👇

이것도 전혀 나쁘지 않지만, 많은 사용 사례에서 우리는 FastAPI Dependencies를 사용하고 싶어 해요. 예를 들어 클라이언트가 로그인했는지 확인하고, path operation의 다른 코드를 실행하기 전에 현재 사용자를 얻기 위해서요.

이 dependency들은 테스트할 때도 아주 유용해요. 쉽게 교체할 수 있어서, 예를 들어 테스트용 새 데이터베이스를 사용하거나, 테스트 전에 데이터를 넣을 수 있으니까요.

그러니 이 세션들을 FastAPI Dependencies를 사용하도록 리팩터링해 볼게요.

FastAPI Dependency 만들기

FastAPI dependency는 아주 단순해요. 값을 반환하는 함수일 뿐이죠.

return 대신 yield를 사용할 수도 있어요. 그 경우 FastAPI는 요청 처리가 끝나면 yield 이후의 코드를 모두 실행하도록 보장해 줘요.

# Code above omitted 👆

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

# Code below omitted 👇

Dependency 사용하기

이제 FastAPI가 dependency를 실행하고 그 값을 path operation에서 얻도록 해 볼게요.

fastapi에서 Depends()를 가져와요. 그런 다음 path operation 함수의 파라미터에서, JSON 본문이나 path 파라미터 등을 얻기 위해 파라미터를 선언했던 것과 같은 방식으로 사용해요.

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

# Code here omitted 👈

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

# Code here omitted 👈

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

# Code below omitted 👇

💡 팁

파라미터들의 그 *(별표)에 대한 팁 하나를 알려줄게요.

여기서 우리는 Depends(get_session)이라는 "기본값"을 가진 session 파라미터를, 기본값이 없는 hero 파라미터 앞에 전달하고 있어요.

Python은 평소에 그것에 대해 불평할 거예요. 하지만 처음 "파라미터"인 *를 사용해서 나머지 모든 파라미터를 "키워드 전용(keyword only)"으로 표시하면 그 문제가 해결돼요.

더 자세한 내용은 FastAPI 문서의 Path Parameters and Numeric Validations - Order the parameters as you need, tricks에서 읽을 수 있어요.

dependency의 값은 한 요청에만 사용돼요. FastAPI는 여러분의 코드를 호출하기 바로 전에 그 dependency를 호출하고, 그 dependency로부터 얻은 값을 전달해 줘요.

만약 yield가 있다면, 응답을 보내는 것이 끝나면 나머지 실행을 계속할 거예요. 세션의 경우, with 블록의 정리 코드를 마저 실행해서 세션을 닫는 등의 일을 하게 돼요.

그런 다음 FastAPI는 다음 요청에 대해 그것을 다시 호출할 거예요.

요청마다 한 번씩 호출되므로, 우리는 원래처럼 요청당 세션 하나를 여전히 얻게 돼요. 여전히 제대로 된 거죠. ✅

그리고 dependency는 yield를 사용할 수 있으므로, FastAPI는 끝나면 yield 이후의 코드(마지막 with 블록의 모든 정리 코드 포함)를 실행하도록 보장해 줘요. 그래서 이것도 제대로 돼요. ✅

with 블록

이 말은 path operation 함수의 메인 코드가 이전 버전의 명시적 with 블록과 동등하게 동작한다는 뜻이에요.

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

# Code here omitted 👈

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

# Code here omitted 👈

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

# Code below omitted 👇

사실, create_hero() 함수 안의 그 코드 블록 전체가 여전히 세션을 위한 with 블록 안에 있다고 생각할 수도 있어요. 왜냐하면 뒤에서 일어나는 일이 대략 그렇기 때문이에요.

하지만 이제 with 블록은 함수 안에 명시적으로 있지 않고, 위의 dependency 안에 있어요:

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


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


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


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


class HeroUpdate(SQLModel):
    name: str | None = None
    secret_name: str | None = None
    age: int | 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, echo=True, connect_args=connect_args)


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


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


app = FastAPI()


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


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate):
    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: Session = Depends(get_session),
    offset: int = 0,
    limit: int = Query(default=100, le=100),
):
    heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
    return heroes


@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(*, session: Session = Depends(get_session), hero_id: int):
    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(
    *, session: Session = Depends(get_session), hero_id: int, hero: HeroUpdate
):
    db_hero = session.get(Hero, hero_id)
    if not db_hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    hero_data = hero.model_dump(exclude_unset=True)
    db_hero.sqlmodel_update(hero_data)
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero


@app.delete("/heroes/{hero_id}")
def delete_hero(*, session: Session = Depends(get_session), hero_id: int):
    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}

나중에 코드를 테스트할 때 이것이 얼마나 유용한지 보게 될 거예요. ✅

Path Operations를 Dependency를 사용하도록 업데이트하기

이제 나머지 path operation들도 새 dependency를 사용하도록 업데이트할 수 있어요.

함수의 파라미터에 이렇게 dependency를 선언하기만 하면 돼요:

session: Session = Depends(get_session)

그리고 나서 이전 세션과 함께 있던 예전 with 블록을 제거해요.

정리

방금 FastAPI dependency를 사용해 데이터베이스 세션을 다루는 방법을 배웠어요. 이건 나중에 코드를 테스트할 때 유용하게 쓰일 거예요.

그리고 FastAPI로 작업하면 할수록 이 dependency들이 큰 도움이 된다는 걸 보게 될 거예요. 권한, 인증, 데이터베이스 세션 같은 리소스 등을 다루는 데 말이죠. 🚀

dependency에 대해 더 배우고 싶다면 FastAPI 문서의 Dependencies 부분을 확인해 보세요.

FastAPI Dependency로 세션 다루기