[Laboratrix] 4 - FastAPI로 CRUD 동작 구현하기

 

Intro

이전 시간에는 Docker Compose로 Backend 개발 환경을 구성했었습니다.

[Laboratrix] 3 - Docker Compose로 backend 개발 환경 구성

 

이번에는 Backend의 기본 동작을 구현해 보겠습니다. Item의 CRUD 동작을 만들어 보려 합니다. 적어도 unittest는 돌릴 수 있는 수준이 되어야겠죠..?

unittest를 할 수 있을 정도의 기능 개발을 진행

 

---

FastAPI 기본 code 작성

FastAPI는 Python 언어를 사용하는 Web Framework중에서 고성능의 Web Framework입니다. 이 Framework를 활용해 보겠습니다.

FastAPI

 

아직 설치하지 않으셨다면 pip로 설치를 진행하시기 바랍니다.

pip install fastapi uvicorn

 

 

추가적을 설치할 package는 SQLalchemy입니다. SQLAlchemy는 Python에서 데이터베이스와의 상호작용을 쉽게 해주는 ORM(Object Relational Mapping) 라이브러리입니다. 이것도 설치합시다.

SQLAlchemy

pip install sqlalchemy

 

 

환경변수를 관리할 load_dotenv도 설치합니다.

pip install python-dotenv

 

 

PostgreSQL DB와 연결하기 위해 psycopg2-binary를 설치합니다.

만약 한 container에 Fastapi와 Postgresql가 같이 설치되어 있으면 psycopg2를 설치하셔도 됩니다.(하지만 이건 container의 단일 책임 원칙과 어긋나는 구조입니다.)

pip install psycopg2-binary

 

 

 

설치가 완료되었으면 코드를 작성해 볼까요?

 

---

Code 작성

 

 

우선 main.py에는 다음과 같이 작성합니다.

from fastapi import FastAPI
from app.api import v1
from dotenv import load_dotenv
from app.db.connection import create_tables, drop_tables


load_dotenv()


app = FastAPI(
    on_startup=[create_tables],
    on_shutdown=[drop_tables]
    )

app.include_router(v1.router, prefix='/api/v1')


if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

 

 

load_dotenv()를 통해 project의 .env 파일을 읽어 환경변수값으로 사용할 수 있습니다.

.env에는 다음과 같은 값을 작성합니다. (DB_PASSWORD는 각자의 값을 작성합니다.)

BACKEND_PORT=8000

DB_TYPE=postgresql
DB_USER=tyoon9781
DB_PASSWORD=...
DB_HOST=local_db
DB_NAME=localdb
DB_PORT=5432

 

 

.env파일을 만든 김에 docker-compose.yml 파일을 .env의 환경변수를 통해 container를 생성할 수 있도록 합시다. 이렇게 하면 민감한 정보를 쉽게 관리할 수 있습니다.

* container name은 환경변수를 통해 생성할 수 없습니다.

services:
  local_backend:
    image: python:3.12.6-slim
    container_name: local_backend
    volumes:
      - .:/app
    working_dir: /app
    ports:
      - "${BACKEND_PORT}:${BACKEND_PORT}"
    depends_on:
      - local_db
    command: tail -f /dev/null

  local_db:
    image: postgres:16.4-bookworm
    container_name: local_db
    environment:
      POSTGRES_USER: ${DB_USER}
      POSTGRES_PASSWORD: ${DB_PASSWORD}
      POSTGRES_DB: ${DB_NAME}
    ports:
      - "${DB_PORT}:${DB_PORT}"

 

 

.env 파일에 backend port도 표기를 했으니 main.py에 적용하도록 하겠습니다.

if __name__ == "__main__":
    import os
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=int(os.getenv('BACKEND_PORT')), reload=True)

 

 

이번에는 app/db폴더를 만들고 app/db/connection.py를 작성합니다. 이 코드는 DB와의 연결을 담당합니다. 

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


DB_TYPE = os.getenv("DB_TYPE")
DB_USER = os.getenv("DB_USER")
DB_PASSWORD = os.getenv("DB_PASSWORD")
DB_HOST = os.getenv("DB_HOST")
DB_NAME = os.getenv("DB_NAME")
DB_PORT = os.getenv("DB_PORT")
DB_URL = f"{DB_TYPE}://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"


engine = create_engine(DB_URL, pool_size=100)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()


def create_tables():
    Base.metadata.create_all(bind=engine)


def drop_tables():
    Base.metadata.drop_all(bind=engine)


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

 

 

이번에는 app/db/models/item.py를 작성합니다. 이 코드는 DB의 ItemModel객체에 대한 table 설계와 Item을 부르를 담당합니다.

from sqlalchemy import Column, Integer, String, DateTime
from datetime import datetime, timezone
from pydantic import BaseModel, ConfigDict
from sqlalchemy.sql import func
from app.db.connection import Base


def utc_now():
    return datetime.now(tz=timezone.utc)


#####################
## Model
#####################
class ItemModel(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, nullable=True)
    created_at = Column(DateTime(timezone=True), default=utc_now)
    updated_at = Column(DateTime(timezone=True), default=utc_now, onupdate=utc_now)


#####################
## Type
#####################
class ItemCreate(BaseModel):
    name: str
    description: str|None = None


class ItemRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    name: str
    description: str|None = None
    created_at: datetime
    updated_at: datetime


class ItemUpdate(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    description: str|None = None


class ItemDelete(BaseModel):
    pass

 

 

이번에는 app/db/crud.py 파일을 만들어 줍시다. DB의 실제 Data의 관리를 맡게 됩니다.

from sqlalchemy.orm import Session
from app.db.models.item import *


def create_item(db: Session, item: ItemCreate) -> ItemModel:
    db_item = ItemModel(**item.model_dump())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item


def get_items(db: Session, skip: int = 0, limit: int = 10) -> list[ItemModel]|None:
    return db.query(ItemModel).offset(skip).limit(limit).all()


def get_item(db: Session, item_id: int) -> ItemModel|None:
    return db.query(ItemModel).filter(ItemModel.id == item_id).first()


def update_item(db: Session, item_id: int, item_update: ItemUpdate) -> ItemModel|None:
    db_item = get_item(db, item_id)
    if db_item is None:
        return None
    
    db_item.description = item_update.description
    db.commit()
    db.refresh(db_item)
    return db_item


def delete_item(db: Session, item_id: int) -> ItemModel|None:
    db_item = get_item(db, item_id)
    if db_item is None:
        return None
    
    db.delete(db_item)
    db.commit()
    return db_item

 

 

이번에는 app/api/v1.py 파일을 만들어 줍시다. 외부의 어떤 request를 받고, 어떻게 처리할지를 정의합니다.

from fastapi import APIRouter, HTTPException, Depends
from typing import List
from sqlalchemy.orm import Session
from app.db import crud
from app.db.models.item import ItemCreate, ItemRead, ItemUpdate, ItemDelete
from app.db.connection import get_db


router = APIRouter()


@router.post("/items/", response_model=ItemRead)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = crud.create_item(db=db, item=item)
    return db_item


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


@router.get("/items/{item_id}", response_model=ItemRead)
def read_item(item_id: int, db: Session = Depends(get_db)):
    item = crud.get_item(db=db, item_id=item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item


@router.put("/items/{item_id}", response_model=ItemRead)
def update_item(item_id: int, item_update: ItemUpdate, db: Session = Depends(get_db)):
    db_item = crud.update_item(db=db, item_id=item_id, item_update=item_update)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item


@router.delete("/items/{item_id}", response_model=ItemDelete)
def delete_item(item_id: int, db: Session = Depends(get_db)):
    db_item = crud.delete_item(db=db, item_id=item_id)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return ItemDelete(id=item_id)

 

 

코드 작성하시느라 고생 많았습니다. (복붙보다는 손수 직접 작성하시는 것을 추천 드립니다)

여기까지 작성이 완료되었으면 이제 backend api를 실행해볼 차례입니다!

 

 

---

CRUD 동작 실행

간단한 동작을 한번 확인해 볼까요? backend container에 fastapi를 실행시키고 chrome에 접속해 아래의 url을 호출해 봅니다.

[GET] http://127.0.0.1:8000/api/v1/items/

 

 

이 url은 item들의 목록을 확인할 수 있는 api입니다. 실행해보면 아래와 같은 결과를 얻을 수 있습니다.

 

 

현재는 DB에 item이 없기 때문에 빈 목록이 출력됩니다. 빈 목록이 출력되지 않도록 item을 추가해 줍시다. item 추가, 삭제 등 api를 수월하게 Test하기 위해 Postman을 설치하도록 하겠습니다.

Postman API Platform | Sign Up for Free

 

 

Postman 설치가 끝났으면 item을 생성해 보도록 하겠습니다

 

Create

아래와 같이 url을 입력하고 POST Method를 설정, Body값을 입력합니다.

[POST] /api/v1/items/
body : 
{
    "name": "testname",
    "description": "testdescription"
}

 

 

POST request를 Send 해보면 아래와 같은 응답을 받을 수 있습니다.

 

 

Read

이제 다시 items를 조회해보면 추가한 Item이 잘 출력되는 것을 확인할 수 있습니다.

[GET] /api/v1/items/

 

 

 

이번에는 2번 item만 조회를 해보도록 하겠습니다.

[GET] /api/v1/items/2

 

 

Update

이번에는 기존의 데이터를 수정해 보도록 하겠습니다. 2번 item의 description을 수정해 보겠습니다.

[PUT] /api/v1/items/2
BODY : {
  "description": "new_description_update"
}

 

 

2번의 item이 변경되었습니다. 확인해 보면 description이 변경된 것을 확인할 수 있습니다.

 

 

Delete

이번에는 Delete를 해보겠습니다. 1번 아이템을 지워보겠습니다.

[DELETE] /api/v1/items/1

지워질 때 return을 정의하지 않아 빈 object가 출력되었다. status 200으로 정상동작이 되었음을 확인할 수 있다.

 

 

제대로 지워졌는지 조회해보면 1번 아이템이 삭제된 것을 확인해 볼 수 있습니다.

 

---

Outro

이번 글에서는 Backend API를 간단하게 구축해봤습니다. Docker compose를 통해 Container 구축과 DB 연결을 보다 쉽게 진행하고 있습니다. 이번 CRUD 동작을 통해 FastAPI와 SQLAlchemy의 궁합이 꽤 좋다는 것을 아셨으면 합니다. 다음 글에서는 UnitTest를 진행해보도록 하겠습니다. 감사합니다.

 

 

 

 

 

* reference 

https://docs.python.org/ko/3.9/library/datetime.html#datetime.datetime.now