본문 바로가기
Web 개발/FAST API (인프런 강의 내용)

2 실습10 ORM 적용 - PATCH API

by yororing 2024. 6. 1.

앞 단계 참조 링크:

00 개요

  • PATCH(업데이트) API에 ORM 적용해주는 실습 진행

01 ToDo 수정하기

1. orm.py에 쓰기

  • is_done의 값을 업데이트/변경 할 새로운 인스턴스 메소드 done(), undone() 정의하기
  • 이렇게 하면 유지보수가 더 용이함 - 반복될 수 있는 작들은 인스턴스로 정의해놓으면 시스템을 관리/통제하기 쉬움
# /c/Users/관리자/Desktop/projects/todos/src/database/orm.py 내용

...

def done(self) -> "ToDo":				# 추가
    # ToDo의 is_done 값을 True로 변경 후 ToDo 반환
    self.is_done = True
    return self
    
def undone(self) -> "ToDo":				# 추가
    # ToDo의 is_done 값을 False로 변경 후 ToDo 반환
    self.is_done = False
    return self

2. repository.py에 쓰기

  • 업데이트/변경한 내용을 DB에 저장하기 위해 update_todo() 함수 정의
  • 정의할 update_todo() 함수는 앞서 정의한 create_todo() 힘수와 동일할 것 
    • 이유: 업데이트된 ToDo 객체를 받아 Session에 추가해주고, Session에 commit하여 DB에 저장 후 DB의 최신 데이터를 잘 가져오는지 확인하기 위해 refresh 함 (Note: update_todo() 함수 내에서 session.refresh()는 필수 아니지만 잘 가져오는지 확인하고자 넣음)
# /c/Users/관리자/Desktop/projects/todos/src/database/repository.py 내용

...
def create_todo(session: Session, todo: ToDo) -> ToDo:		# 원래 있었음
    session.add(instance=todo)
    session.commit()
    session.refresh(instance=todo)
    return todo

def update_todo(session: Session, todo: ToDo) -> ToDo:		# 추가
    session.add(instance=todo)
    session.commit() # DB save
    session.refresh(instance=todo)
    return todo
  • create_todo()와 update_todo()의 코드 내용이 동일하지만 repository 패의 특성상 데이터를 관리하는 부분을 명시적으로 분리해주는 것이 더 낫기 때문에 update_todo() 함수 생성하는 것 

3. main.py에 쓰기

# /c/Users/관리자/Desktop/projects/todos/src/main.py 내용
from fastapi import FastAPI, Body, HTTPException, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
from typing import List	

from database.connection import get_db	
from database.repository import get_todos, get_todo_by_todo_id, create_todo, updat_todo # 추가
from database.orm import ToDo
from schema.response import ToDoSchema, ToDoListSchema
from schema.request import CreateToDoRequest

...
# # PATCH Method 사용하여 is_done 값 수정 API
# @app.patch("/todos/{todo_id}", status_code=200)
# def update_todo_handler(
#    todo_id: int,
#    is_done: bool = Body(..., embed=True)
#    ):
#    todo = todo_data.get(todo_id)
#    if todo:
#        todo["is_done"] = is_done
#        return todo
#    raise HTTPException(status_code=404, detail="ToDo Not Found")

# 위 내용을 아래와 같이 수정

# PATCH Method 사용하여 is_done 값 수정 API
@app.patch("/todos/{todo_id}", status_code=200)
def update_todo_handler(
    todo_id: int,
    is_done: bool = Body(..., embed=True)
    session: Session = Depends(get_db),			# 추가
    ):
    todo: ToDo | None = get_todo_by_todo_id(session=session, todo_id=todo_id)
    if todo: # todo가 존재하면
        # 업데이트/변경
        todo.done() if is_done else todo.undone()		# orm.py에 추가된 내용 참조
        							# is_done값이 True라면 todo.done() 실행, False라면 tod.undone() 실행
        todo: ToDo = update_todo(session=session, todo=todo) 	# repository.py에 추가된 내용 참조
        return ToDoSchema.from_orm(todo)
    # todo가 존재하지 않으면 에러 발생
    raise HTTPException(status_code=404, detail="ToDo Not Found")

 

4. 실습으로 확인 - SwaggerUI 문서에서

0) 실습 환경 준비

  • 가상환경 활성화, docker 컨테이너 시작, 경로 src로 이동, uvicorn 실행
# 가상환경 활성화
$ . /c/Users/관리자/Desktop/projects/todos/Scripts/activate

# docker 컨테이너 목록 및 상태 확인
(todos)$ docker ps -a		
CONTAINER ID   IMAGE                             COMMAND                   CREATED       STATUS                    PORTS     NAMES
0c3283e97292   mysql:8.0                         "docker-entrypoint.s…"   2 weeks ago   Exited (0) 24 hours ago             todos
2597a124c27d   docker/welcome-to-docker:latest   "/docker-entrypoint.…"   2 weeks ago   Exited (0) 2 weeks ago              welcome-to-docker

# todos 컨테이너 시작하기
(todos)$ docker start todos
todos

# src 경로로 이동
(todos)$ cd /c/Users/관리자/Desktop/projects/todos/src

# uvicorn 실행
(todos)$ uvicorn main:app --reload
INFO:     Will watch for changes in these directories: ['C:\\Users\\관리자\\Desktop\\projects\\todos\\src']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [8444] using WatchFiles
INFO:     Started server process [6332]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

1) 브라우저에 SwaggerUI 문서 띄우기

  • http://127.0.0.1:8000/docs 입력

 

  • todos 전체 조회 하기: GET /todos Get Todo Handler 클릭 > Try it out > Execute 클릭하여 전체 ToDo 조회해서 업데이트 하고자 하는 todo 고르기 (id=3의 is_done=False를 is_done=True로 업데이트 시켜주기)

  • PATCH /todos/{todo_id} Update Todo Handler 클릭 > Parameters 아래 todo_id 값에 3 입력 > Request body 아래 다음을 입력
{
    "is_done": true
}

  • Execute 클릭
  • Responses 아래 Server response로 200 응답이 오며 id=3 ToDo가 true로 업데이트 된 것을 확인

  • 만약 존재하지 않는 id=12으로 업데이트 요청하면 Server response로 404 (에러) 응답이 올 것

 

  • 존재하지 않는 id=12으로 업데이트 요청하여 Server response로 404 (에러) 응답이 온 것을 확인 

이때까지의 코드들:

  • /c/Users/관리자/Desktop/projects/todos/src/main.py
# /c/Users/관리자/Desktop/projects/todos/src/main.py 내용
from fastapi import FastAPI, Body, HTTPException, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
from typing import List	

from database.connection import get_db	
from database.repository import get_todos, get_todo_by_todo_id, create_todo, update_todo
from database.orm import ToDo
from schema.response import ToDoSchema, ToDoListSchema
from schema.request import CreateToDoRequest

app = FastAPI()

# 첫 화면 API
@app.get("/")
def health_check_handler():
    return {"ping": "pong"}


# 데이터베이스 역할하는 딕셔너리 생성
todo_data = {
    1: {
        "id": 1,
        "content": "실전! FastAPI 섹션 0 수강",
        "is_done": True,
    },
    2: {
        "id": 2,
        "content": "실전! FastAPI 섹션 1 수강",
        "is_done": False,
    },
    3: {
        "id": 3,
        "content": "실전! FastAPI 섹션 2 수강",
        "is_done": False,
    },
}

# GET Method 사용하여 전체 조회 API - 딕셔너리 참조
# @app.get("/todos", status_code=200)
# def get_todos_handler(order: str | None = None):
#     rt = list(todo_data.values())
#     if order and order == "DESC":
#        return rt[::-1]
#     return rt

# GET Method 사용하여 전체 조회 API - DB 참조
@app.get("/todos", status_code=200)
def get_todos_handler(
    order: str | None = None,
    session: Session = Depends(get_db),
) -> ToDoListSchema:
    todos: List[ToDo] = get_todos(session=session)
    if order and order == "DESC":
#       return todos[::-1]
        return ToDoListSchema(
            todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
        )
#   return todos
    return ToDoListSchema(
        todos=[ToDoSchema.from_orm(todo) for todo in todos]
    )

# GET Method 사용하여 단일 조회 API
@app.get("/todos/{todo_id}", status_code=200)
def get_todo_handler(
    todo_id: int,
    session: Session=Depends(get_db),
    ) -> ToDoSchema:
    todo:ToDo | None = get_todo_by_todo_id(session=session, todo_id = todo_id)
    if todo:
        return ToDoSchema.from_orm(todo)
    raise HTTPException(status_code=404, detail="ToDo Not Fount")

# POST Medthod 사용하여 todo 생성 API
@app.post("/todos", status_code=201)
def create_todo_handler(
    request: CreateToDoRequest,
    session: Session = Depends(get_db),
) -> ToDoSchema:
    todo: ToDo = ToDo.create(request=request)            # id=None
    todo: ToDo = create_todo(session=session, todo=todo) # id=int
    return ToDoSchema.from_orm(todo)

# PATCH Method 사용하여 is_done 값 수정 API
@app.patch("/todos/{todo_id}", status_code=200)
def update_todo_handler(
    todo_id: int,
    is_done: bool = Body(..., embed=True),
    session: Session = Depends(get_db),
    ):
    todo:ToDo | None = get_todo_by_todo_id(session=session, todo_id = todo_id)
    if todo:
        # update - is_done값이 True이면 todo.done() 실행, False이면 todo.undone() 실행
        todo.done() if is_done else todo.undone()
        todo: ToDo = update_todo(session=session, todo=todo)
        return ToDoSchema.from_orm(todo)
    raise HTTPException(status_code=404, detail="ToDo Not Fount")

# DELETE Method 사용하여 todo 아이템 삭제 API
@app.delete("/todos/{todo_id}", status_code=204)
def delete_todo_handler(todo_id: int):
    todo = todo_data.pop(todo_id, None)
    if todo:
        return
    raise HTTPException(status_code=404, detail="ToDo Not Found")
  • /c/Users/관리자/Desktop/projects/todos/src/database/orm.py
# /c/Users/관리자/Desktop/projects/todos/src/database/orm.py 내용
from sqlalchemy import Boolean, Column, Integer, String
from sqlalchemy.orm import declarative_base
from schema.request import CreateToDoRequest

Base = declarative_base()

# ToDo 클래스 모델링 한 것
class ToDo(Base):
    __tablename__ = 'todo'
    
    id = Column(Integer, primary_key=True, index=True)
    contents = Column(String(256), nullable=False)
    is_done = Column(Boolean, nullable=False)
    
    def __repr__(self):
        return f"ToDo(id={self.id}, contents={self.contents}, is_done={self.is_done})"
    
    @classmethod
    def create(cls, request: CreateToDoRequest) -> "ToDo":
        return cls(
            # id값은 DB에 의해 자동으로 결정 되서 별도로 지정안해줌
            contents=request.contents,
            is_done=request.is_done,
        )
        
    def done(self) -> "ToDo":
        # ToDo의 is_done 값을 True로 변경 후 ToDo 반환
        self.is_done = True
        return self
    
    def undone(self) -> "ToDo":
        # ToDo의 is_done 값을 False로 변경 후 ToDo 반환
        self.is_done = False
        return self
  • /c/Users/관리자/Desktop/projects/todos/src/database/connection.py
# /c/Users/관리자/Desktop/projects/todos/src/database/connection.py 내용
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "mysql+pymysql://root:todos@127.0.0.1:3306/todos"

engine = create_engine(DATABASE_URL, echo=True)
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    session = SessionFactory()
    try:
        yield session
    finally:
        session.close()
  • /c/Users/관리자/Desktop/projects/todos/src/database/repository.py
# /c/Users/관리자/Desktop/projects/todos/src/database/repository.py 내용
from sqlalchemy import select
from sqlalchemy.orm import Session
from typing import List

from database.orm import ToDo

def get_todos(session: Session) -> List[ToDo]:
    return list(session.scalars(select(ToDo)))

def get_todo_by_todo_id(session: Session, todo_id: int) -> ToDo | None:
    return session.scalar(select(ToDo).where(ToDo.id == todo_id))

def create_todo(session: Session, todo: ToDo) -> ToDo:
    session.add(instance=todo)
    session.commit()
    session.refresh(instance=todo)
    return todo

def update_todo(session: Session, todo: ToDo) -> ToDo:
    session.add(instance=todo)
    session.commit()
    session.refresh(instance=todo)
    return todo
  • /c/Users/관리자/Desktop/projects/todos/src/schema/response.py
# /c/Users/관리자/Desktop/projects/todos/src/schema/response.py 내용
from pydantic import BaseModel
from typing import List

class ToDoSchema(BaseModel):
    id: int
    contents: str
    is_done: bool

    class Config:
        from_attributes = True

class ToDoListSchema(BaseModel):
    todos: List[ToDoSchema]
  • /c/Users/관리자/Desktop/projects/todos/src/schema/request.py
# /c/Users/관리자/Desktop/projects/todos/src/schema/request.py 내용

from pydantic import BaseModel

class CreateToDoRequest(BaseModel):
    contents: str
    is_done: bool