FastAPI 찍먹 3편

✅ 아래 내용들에 대해서 알아보자

- 프로젝트 세팅 및 실습
- 테스트

 

 

공식문서를 보면서 FastAPI & Sqlalchemy를 같이 사용해 보자!

 

프로젝트 세팅

 프로젝트 폴더 구조는 sql_app이라는 폴더가 있고 그 밑에 각각 역할을 하는 파일이 존재한다.

 

프로젝트 구조

 

각 파일들을 하나씩 살펴보자!

 

1. crud.py - DB와 통신하여 crud 작업을 하는 역할(persistence layer, 스프링의 repository 클래스

from sqlalchemy.orm import Session
from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


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 = 10):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notReallyhashed"
    db_user = models.User(
        email=user.email,
        hashed_password=fake_hashed_password,
    )
    db.add(db_user)
    db.commit()  # db 커밋
    db.refresh(db_user)  # 커밋 후 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

 

 

2. database.py - DB connection 설정 파일

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker


# SQLAlchemy db 연결(여기서는 SQLite) 속성 설정
SQLALCHEMY_DATABASE_URL = "sqllite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgreserver/db"

# check_same_thread는 SQLite에서만 필요
engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},
)


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

# 이 클래스를 상속하여 각 ORM 생성
Base = declarative_base()

 

 

3. main.py - 파이썬 시작 모듈 및 api 모듈 (스프링의 controller 클래스)

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


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

models.Base.metadata.create_all(bind=engine)


app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


db = Depends(get_db)


# 유저 생성 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="이메일 중복!")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def get_users(skip: int = 0, limit: int = 100):
    users = crud.get_users(db=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):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items")
def get_items(skip: int = 0, limit: int = 100):
    items: list[models.Item] = crud.get_items(
        db=db,
        skip=skip,
        limit=limit,
    )
    return items

 

 

4. models.py - 엔티티 집합(스프링의 엔티티 클래스)

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

# SQLAlchemy 및 기타 많은 항목은 기본적으로 "지연 로딩"입니다.


class User(Base):
    __tablename__ = "users"  # db에서 사용할 테이블 이름을 sqlalchemy에 알려줌

    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")

 

 

5. schemas.py - 정확하진 않지만 DTO 클래스들 모은 느낌...?

from pydantic import BaseModel


# DTO 클래스 같은데..?


class ItemBase(BaseModel):
    title: str
    description: str | None = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


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

 

 

테스트

 

---

참고자료

• https://fastapi.tiangolo.com/ko/tutorial/sql-databases/