FastAPI의 데이터베이스 연동 쉽게 이해하기

소개

웹 애플리케이션에서 데이터베이스는 필수적입니다. 이번 글에서는 FastAPI에서 SQLAlchemy를 사용하여 데이터베이스를 연동하는 방법을 예제와 함께 상세히 알아보겠습니다. 특히 온라인 쇼핑몰의 상품 관리 시스템을 예제로 사용하여 실제 활용 방법을 이해하기 쉽게 설명하겠습니다.

데이터베이스 설정하기

먼저 필요한 패키지를 설치합니다.

pip install sqlalchemy psycopg2-binary
# sqlalchemy: 데이터베이스 작업을 위한 ORM
# psycopg2-binary: PostgreSQL 연결을 위한 드라이버

데이터베이스 연결을 설정합니다:

# database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 데이터베이스 URL 설정
# format: "postgresql://사용자이름:비밀번호@호스트/데이터베이스이름"
DATABASE_URL = "postgresql://myuser:mypassword@localhost/myshop"

# 데이터베이스 엔진 생성
engine = create_engine(DATABASE_URL)

# 세션 생성 클래스 설정
# autocommit=False: 수동으로 커밋해야 변경사항이 저장됨
# autoflush=False: 수동으로 flush 해야 데이터가 DB에 반영됨
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 모델 기본 클래스 생성
Base = declarative_base()

데이터 모델 정의하기

온라인 쇼핑몰의 상품 데이터 모델을 정의해보겠습니다:

# models.py
from sqlalchemy import Column, Integer, String, Float, Boolean
from database import Base

class Product(Base):
    # 테이블 이름 지정
    __tablename__ = "products"

    # 컬럼 정의
    id = Column(Integer, primary_key=True, index=True)
    # index=True는 검색 속도를 높이기 위한 인덱스 생성
    name = Column(String, index=True)        # 상품명
    description = Column(String)             # 상품 설명
    price = Column(Float)                    # 가격
    is_available = Column(Boolean, default=True)  # 판매 여부

    def __repr__(self):
        # 객체를 문자열로 표현할 때 사용
        return f"Product(id={self.id}, name={self.name}, price={self.price})"

Pydantic 모델 정의하기

API 요청과 응답의 데이터 형식을 정의합니다:

# schemas.py
from pydantic import BaseModel

class ProductBase(BaseModel):
    name: str
    description: str | None = None  # 선택적 필드
    price: float
    is_available: bool = True

class ProductCreate(ProductBase):
    # 생성 시 사용할 스키마
    pass

class Product(ProductBase):
    # DB 조회 시 사용할 스키마
    id: int

    class Config:
        # ORM 모델을 Pydantic 모델로 자동 변환
        orm_mode = True

데이터베이스 작업 함수 만들기

CRUD(Create, Read, Update, Delete) 작업을 위한 함수들을 만듭니다:

# crud.py
from sqlalchemy.orm import Session
import models
import schemas

def create_product(db: Session, product: schemas.ProductCreate):
    """
    새로운 상품을 데이터베이스에 추가합니다.

    사용 예시:
    new_product = schemas.ProductCreate(
        name="맥북 프로 16인치",
        price=3500000,
        description="최신 프로세서 탑재"
    )
    db_product = create_product(db, new_product)
    """
    # 데이터베이스 모델 인스턴스 생성
    db_product = models.Product(**product.dict())
    # 세션에 추가
    db.add(db_product)
    # 변경사항 저장
    db.commit()
    # 저장된 데이터 새로고침
    db.refresh(db_product)
    return db_product

def get_product(db: Session, product_id: int):
    """
    ID로 상품을 조회합니다.

    사용 예시:
    product = get_product(db, product_id=1)
    """
    return db.query(models.Product).filter(models.Product.id == product_id).first()

def get_products(db: Session, skip: int = 0, limit: int = 100):
    """
    여러 상품을 조회합니다.

    사용 예시:
    products = get_products(db, skip=0, limit=10)  # 첫 10개 상품 조회
    """
    return db.query(models.Product).offset(skip).limit(limit).all()

def update_product(db: Session, product_id: int, product: schemas.ProductCreate):
    """
    상품 정보를 업데이트합니다.

    사용 예시:
    updated_info = schemas.ProductCreate(
        name="맥북 프로 14인치",
        price=2900000
    )
    updated_product = update_product(db, product_id=1, product=updated_info)
    """
    db_product = get_product(db, product_id)
    if db_product:
        # 모델의 각 필드 업데이트
        for key, value in product.dict().items():
            setattr(db_product, key, value)
        db.commit()
        db.refresh(db_product)
    return db_product

API 엔드포인트 구현하기

이제 실제 API 엔드포인트를 만들어봅시다:

# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
import crud, models, schemas
from database import SessionLocal, engine

# 데이터베이스 테이블 생성
models.Base.metadata.create_all(bind=engine)

app = FastAPI()

# 데이터베이스 세션 의존성
def get_db():
    """
    데이터베이스 세션을 생성하고 관리합니다.
    with 문처럼 사용이 끝나면 자동으로 세션을 닫습니다.
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/products/", response_model=schemas.Product)
def create_product(product: schemas.ProductCreate, db: Session = Depends(get_db)):
    """
    새로운 상품을 생성합니다.

    요청 예시:
    POST /products/
    {
        "name": "아이폰 15 Pro",
        "description": "최신 스마트폰",
        "price": 1500000
    }
    """
    return crud.create_product(db=db, product=product)

@app.get("/products/", response_model=List[schemas.Product])
def read_products(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    """
    상품 목록을 조회합니다.

    요청 예시:
    GET /products/?skip=0&limit=10
    """
    products = crud.get_products(db, skip=skip, limit=limit)
    return products

@app.get("/products/{product_id}", response_model=schemas.Product)
def read_product(product_id: int, db: Session = Depends(get_db)):
    """
    특정 상품을 조회합니다.

    요청 예시:
    GET /products/1
    """
    product = crud.get_product(db, product_id=product_id)
    if product is None:
        raise HTTPException(status_code=404, detail="상품을 찾을 수 없습니다")
    return product

API 테스트하기

# 터미널에서 서버 실행
uvicorn main:app --reload

# 다른 터미널에서 curl로 테스트
# 상품 생성
curl -X POST "http://localhost:8000/products/" \
     -H "Content-Type: application/json" \
     -d '{"name": "맥북 프로 16인치", "price": 3500000, "description": "최신 프로세서 탑재"}'

# 상품 조회
curl "http://localhost:8000/products/1"

# 상품 목록 조회
curl "http://localhost:8000/products/?skip=0&limit=10"

마치며

이번 글에서는 FastAPI와 SQLAlchemy를 사용하여 데이터베이스를 연동하는 방법을 알아보았습니다. 실제 온라인 쇼핑몰의 상품 관리 시스템을 예제로 사용하여 데이터 모델 정의부터 API 엔드포인트 구현까지 전 과정을 살펴보았습니다.