[FastAPI] FastAPI (2) - PostgreSQL CRUD API 만들기

들어가며

지난 시간에 FastAPI 프로젝트를 구축하고 실행까지 진행했다.

이번에는 PostgreSQL DB를 연동하고 간단한 CRUD API를 만들 것이다. 

(FastAPI Document가 넘넘 잘 되어있어서 많이 참고했다!)

API 동작 결과는 Postman 으로 확인한다.

 

패키지 설치

• fastapi 에서는 orm으로 sqlalchemy 를 사용한다고 한다. (관련 패키지 설치)

$ poetry add psycopg2
$ poetry add psycopg2-binary
$ poetry add sqlalchemy

 

프로젝트 구조

• core > db 폴더 추가

health
├── README.md
├── app
│   ├── __init__.py
│   ├── core
│   │   ├── __init__.py
│   │   ├── config.py
│   │   └── db
│   │       ├── __init__.py
│   │       ├── base.py
│   │       ├── base_class.py
│   │       ├── models.py
│   │       └── schemas.py
│   └── main.py
├── poetry.lock
├── pyproject.toml
├── requirements.txt
└── tests
    └── __init__.py

 

DataBase 연결

# core > db > base.py

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

SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

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

 

테이블 구조 정의

# core > db > models.py

from sqlalchemy import Column, Integer, String, CHAR
from .database import Base


class Users(Base):
    __tablename__ = "health_users"

    id = Column(Integer, primary_key=True, index=True)
    nickname = Column(String(40))
    gender = Column(CHAR(2))
    age = Column(Integer(4))

 

테이블 속성 타입 정의

• API에서 데이터를 반환하고 읽을 때 사용할 모델(스키마) 정의

# core > db > schemas.py

from pydantic import BaseModel


class UserBase(BaseModel):
    nickname: str
    gender: str
    age: int


class User(UserBase):
    id: int

    class Config:
        orm_mode = True

 

CRUD

# core > db > crud.py

def create_user(session: Session, user: schemas.UserBase) -> Users:
    db_user = Users(
        nickname=user.nickname,
        gender=user.gender,
        age=user.age,
    )
    session.add(db_user)
    session.commit()
    session.refresh(db_user)

    return db_user


def update_user_info(
    session: Session, user_id: int, info_update: schemas.UserBase
) -> Users:
    user_info = session.query(models.Users).get(user_id)

    if user_info is None:
        raise HTTPException(status_code=404, detail="ID에 해당하는 User가 없습니다.")

    user_info.nickname = info_update.nickname
    user_info.gender = info_update.gender
    user_info.age = info_update.age

    session.commit()
    session.refresh(user_info)

    return user_info

 

CURD API 정의

# main.py

@app.post("/users/", response_model=schemas.User)
def create_user(user_info: schemas.UserBase, session: Session = Depends(get_db)):
    db_user = (
        session.query(models.Users)
        .filter(models.Users.nickname == user_info.nickname)
        .first()
    )

    if db_user:
        raise HTTPException(status_code=400, detail="Nickname already registered")
    return crud.create_user(session, user_info)


@app.get(path="/users/{user_id}", response_model=schemas.User)
def get_user(user_id: int, session: Session = Depends(get_db)):
    user = session.query(models.Users).filter(models.Users.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="ID에 해당하는 User가 없습니다.")
    return user


@app.put(path="/users/{user_id}", response_model=schemas.User)
def put_user(
    user_id: int, new_info: schemas.UserBase, session: Session = Depends(get_db)
):
    user_info = crud.update_user_info(session, user_id, new_info)
    return user_info


@app.delete(path="/users/{user_id}")
def delete_user(user_id: int, session: Session = Depends(get_db)):
    user_info = session.query(models.Users).get(user_id)

    if user_info is None:
        raise HTTPException(status_code=404, detail="ID에 해당하는 User가 없습니다.")

    session.delete(user_info)
    session.commit()

    return user_info