Colin's Blog
[python] FastAPI로 데이터베이스와의 상호작용 구현하기

FastAPI는 빠르고 사용하기 쉬운 API를 개발할 수 있도록 도와주는 웹 프레임워크입니다. 이번 글에서는 FastAPI를 사용하여 데이터베이스와의 상호작용을 구현하는 방법에 대해 살펴보겠습니다.

데이터베이스 설정

FastAPI에서 데이터베이스를 사용하기 위해서는 먼저 해당 데이터베이스에 연결하는 설정을 해주어야 합니다. 대표적으로 SQLAlchemy나 databases를 사용하여 데이터베이스에 연결할 수 있습니다.

# SQLAlchemy를 사용하는 경우
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)

# databases를 사용하는 경우
import databases

DATABASE_URL = "sqlite:///./test.db"
database = databases.Database(DATABASE_URL)

위 코드에서는 SQLAlchemy를 사용하는 경우 SQLite 데이터베이스에 연결하는 설정을 보여주고, databases를 사용하는 경우에도 동일하게 SQLite 데이터베이스에 연결하는 설정을 보여주었습니다.

모델 정의

다음으로는 데이터베이스에서 사용할 모델을 정의해야 합니다. 이는 데이터베이스 테이블과 매핑되는 클래스로, Pydantic 모델을 사용하여 정의할 수 있습니다.

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, index=True)

위 코드에서는 SQLAlchemy 모델을 정의하는 방법을 보여주었습니다.

데이터베이스 세션과 라우터 설정

데이터베이스에 연결하고 모델을 정의한 후에는 FastAPI 앱에서 이를 사용할 수 있도록 데이터베이스 세션과 라우터를 설정해주어야 합니다.

# SQLAlchemy를 사용하는 경우
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# FastAPI 앱과 데이터베이스 연결
@app.on_event("startup")
async def startup():
    database.connect()
    Base.metadata.create_all(bind=engine)

@app.on_event("shutdown")
async def shutdown():
    database.disconnect()

위 코드에서는 먼저 SQLAlchemy를 사용하는 경우 데이터베이스 세션을 설정하고, FastAPI 앱과 데이터베이스를 연결하는 부분을 보여주었습니다.

데이터베이스 CRUD 라우터 구현

마지막으로, 데이터베이스의 CRUD(Create, Read, Update, Delete) 기능을 수행하는 라우터를 구현해야 합니다.

@app.post("/items/", response_model=schemas.Item)
async def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
    db_item = models.Item(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

@app.get("/items/{item_id}", response_model=schemas.Item)
async def read_item(item_id: int, db: Session = Depends(get_db)):
    return db.query(models.Item).filter(models.Item.id == item_id).first()

@app.put("/items/{item_id}", response_model=schemas.Item)
async def update_item(item_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)):
    db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
    # update 로직 추가
    db.commit()
    return db_item

@app.delete("/items/{item_id}", response_model=schemas.Item)
async def delete_item(item_id: int, db: Session = Depends(get_db)):
    db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
    db.delete(db_item)
    db.commit()
    return db_item

위 코드에서는 각 CRUD 기능에 해당하는 라우터를 구현하는 부분을 보여주었습니다.

이렇게 FastAPI를 사용하여 데이터베이스와의 상호작용을 구현할 수 있습니다. 데이터베이스에 연결하고 모델을 정의한 후에는 원하는 데이터베이스 작업을 수행하는 라우터를 구현하여 API를 완성할 수 있습니다.

더 자세한 내용은 FastAPI 공식 문서를 참고하시기 바랍니다.

© 2018 Colin. All rights reserved.