[FastAPI] FastAPI + SQL 사용하기

Jaeyeon Kim·2023년 6월 29일

FastAPI

목록 보기

3/4

FastAPI의 공식 예제를 참고하여 작성하였다.

예시 코드는 깃허브 레포지토리에 업로드 되어 있다.

꼭 관계형 DB를 사용할 것을 요구하진 않지만, SQLAlchemy 라이브러리를 통해서 SQL을 쉽게 사용할 수 있다.
SQLAlchemy에서 지원하는 DB는 아래와 같다.

PostgreSQL
MySQL
SQLite
Oracle
Microsoft SQL Server, etc

아래 예제는 SQLite의 사용 예제이다.

파일 구조

📦sql_app
 ┣ 📜__init__.py
 ┣ 📜crud.py
 ┣ 📜database.py
 ┣ 📜main.py
 ┣ 📜models.py
 ┗ 📜schemas.py

SQLAlchemy

설치하기

pip install sqlalchemy

DB 생성하기

database.py 파일을 아래와 같이 작성한다.

# 필요한 라이브러리 import하기
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# SQLAlchemy 사용할 DB URL 생성하기
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

# SQLAlchemy engine 생성하기
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

# DB 세션 생성하기
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base class 생성하기
Base = declarative_base()

SQLAlchemy models 생성하기

models.py는 아래와 같다.

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

# database.py에서 생성한 Base import
from .database import Base


# Base를 상속 받아 SQLAlchemy model 생성
class User(Base):
    # 해당 모델이 사용할 table 이름 지정
    __tablename__ = "users"

    # Model의 attribute(column) 생성 -> "="를 사용하여 속성을 정의
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    # 다른 테이블과의 관계 생성
    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

Pydantic models 생성하기

pydantic model은 "스키마"(유효한 데이터 모양)을 어느 정도 정의한다.

from typing import List, Union

from pydantic import BaseModel


# pydantic 라이브러리의 BaseModel 클래스를 상속 받아 ItemBase 생성 -> ":"를 사용하여 속성을 정의
class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None


class ItemCreate(ItemBase):
    pass


# API에서 데이터를 읽을 때/반환할 때 사용될 모델
class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


# UserBase 생성
class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []

    class Config:
        orm_mode = True

CRUD 유틸리티 생성하기

CRUD란?
대부분의 컴퓨터 소프트웨어가 가지는 기본적인 데이터 처리 기능인
Create(생성), Read(읽기), Update(갱신), Delete(삭제)를 묶어서 일컫는 말이다.

from sqlalchemy.orm import Session

# 기존에 생성한 모델과 스키마 불러오기
from . import models, schemas


# 데이터 읽기 - ID로 사용자 불러오기
def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


# 데이터 읽기 - Email로 사용자 불러오기
def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


# 데이터 읽기 - 여러 사용자 불러오기
def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


# 데이터 생성하기
def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"

    # SQLAlchemy 모델 인스턴스 만들기
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)  # DB에 해당 인스턴스 추가하기
    db.commit()  # DB의 변경 사항 저장하기
    db.refresh(db_user)  # 생성된 ID와 같은 DB의 새 데이터를 포함하도록 새로고침
    return db_user


# 데이터 읽기 - 여러 항목 읽어오기
def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

main 앱 살펴보기

from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# 종속성 만들기 : 요청 당 독립적인 데이터베이스 세션/연결이 필요하고 요청이 완료되면 닫음
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# crud.py에서 생성한 crud 유틸리티를 활용하여 API 작성
@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items