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

2 실습8 ORM 적용 - Refactoring

by yororing 2024. 6. 1.

앞 단계 참조 링크:

00 개요

  • ToDo를 생성하는 API에 ORM 적용하기 전 Refactoring하기
  • 현재 main.py에 있는 request(class CreateToDoRequest(BaseModel))은 database 폴더 안에 있는 것이 아니라 schema 폴더에 있어야 됨
  • Class 이름 컨벤션이 안 맞으면 컨벤션에 맞춰야됨
  • 이 request class를 이동하는 refactoring 실습 진행

01 Refactoring

1. schema 폴더 안에 request.py 파일 생성

# touch /c/Users/관리자/Desktop/projects/todos/src/schema/request.py

 

2. request 클래스에 대하여 refactoring 진행

  • 목적: class CreateToDoRequest(BaseModel)을 다른 파일로 옮기기
  • Pycharm에서 가능
    • 'class CreateToDoRequest(BaseModel)'에서 이름 부분 ('CreateToDoRequest')를 하이라이트 > 우클릭 > Refactor > Move... 클릭
    • Move class main.CreateToDoRequest 아래 To:에 옮기고 싶은 경로를 지정해줌
      • '/c/Users/관리자/Desktop/projects/todos/src/schema/request.py' 입력 > Refactor 클릭
  • Move 기능을 이용했기 때문에 이 클래스를 원하는 파일로 옮겨줌: 자동으로 Pycharm이 연관된 import 도 작성/옮겨 줌

2.1. 수동으로 진행

1) request.py 파일에 입력

# /c/Users/관리자/Desktop/projects/todos/src/schema/request.py 내용

from pydantic import BaseModel

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

 

2) main.py 파일 안 위의 내용을 제거 후 request를 import하는 코드 추가

# /c/Users/관리자/Desktop/projects/todos/src/main.py 내용
...
from schema.request import CreateToDoRequest
...

 

3.  /c/Users/관리자/Desktop/projects/todos/src/schema/response.py  Class 이름 수정

  • 목적: class ListToDoResponse(BaseModel): 의 이름 컨벤션이 안 맞아서 이름 수정하기 (ToDoListSchema로 변경)
  • Pycharm에서 가능
    • class ListToDoResponse(BaseModel): 에서 ListToDoResponse를 하이라이트 > 우클릭 > Refactor > Rename 클릭
    • response.py 파일에서 'ListToDoResponse'를 지우고 'ToDoListSchema' 입력 후 Enter 누르기
  • rename 기능을 이용했기 때문에 이 클래스를 사용하는 부분의 모든 이름도 자동으로 Pycharm이 찾아서 변환해줌

 

 

Pycharm 사용 꿀팁:

  • Command + Class이름 누르면 클래스가 정의된 부분으로 이동
  • Refactoring

 

이때까지의 코드들:

  • /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
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):
#     todo = todo_data.get(todo_id)
#     if todo:
#         return todo
#     raise HTTPException(status_code=404, detail="ToD Not Found")

@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_todos_handler(request: CreateToDoRequest):
    todo_data[request.id] = request.dict()
    return todo_data[request.id]

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

# 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

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})"
  • /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))
  • /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:
        orm_mode = 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):
    id: int
    contents: str
    is_done: bool