[FastAPI] Repository 패턴 및 orm 적용

✔️ Session 설정

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

 

✔️ Repository 패턴

database > repository

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

 

Session 을 이용해서 get_todos 함수가 호출될 시에 ToDo Table 의 데이터를 전부 list 로 가져올 수 있도록 로직을 짜주었다.

 

✔️ main.py 적용

main.py

# 할 일 전체 조회
@app.get("/todos", status_code=200)
def get_todos_handler(
        order: str | None = None,
        session: Session = Depends(get_db),
):
    todos: List[ToDo] = get_todos(session=session)

    if order == "DESC":
        return todos[::-1]
    return todos

 

---

💡HTTP Response 설정

schema > response

from pydantic import BaseModel
from typing import List

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

    class Config:
        from_attributes = True

class ListToDoResponse(BaseModel):
    todos: List[ToDoSchema]

 

main,py

# 할 일 전체 조회
@app.get("/todos", status_code=200)
def get_todos_handler(
        order: str | None = None,
        session: Session = Depends(get_db),
):
    todos: List[ToDo] = get_todos(session=session)

    if order == "DESC":
        return ListToDoResponse(
        todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
    )
    return ListToDoResponse(
        todos=[ToDoSchema.from_orm(todo) for todo in todos]
    )

# 할 일 단일 조회
@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 Found")

 

---

💡ORM 적용 ver

repository.py

from sqlalchemy import select, delete
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() # db 저장
    session.refresh(instance=todo) # id 를 가져오기 위함
    return todo

def update_todo(session: Session, todo: ToDo) -> ToDo:
    session.add(instance=todo)
    session.commit() # db 저장
    session.refresh(instance=todo) # id 를 가져오기 위함
    return todo

def delete_todo(session: Session, todo_id:int) -> None:
    session.execute(delete(ToDo).where(ToDo.id == todo_id))
    session.commit()

 

main.py

from fastapi import FastAPI, Body, HTTPException, Depends
from schema.request import CreateTodoRequest
from schema.response import ToDoListSchema, ToDoSchema
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, delete_todo

from database.orm import ToDo

app = FastAPI()

@app.get("/")
def health_check_handler():
    return {"ping":"pong"}


# 할 일 전체 조회
@app.get("/todos", status_code=200)
def get_todos_handler(
        order: str | None = None,
        session: Session = Depends(get_db),
):
    todos: List[ToDo] = get_todos(session=session)

    if order == "DESC":
        return ToDoListSchema(
        todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
    )
    return ToDoListSchema(
        todos=[ToDoSchema.from_orm(todo) for todo in todos]
    )

# 할 일 단일 조회
@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 Found")


# 할 일 생성
@app.post("/todos", status_code=201)
def create_todo_handler(
        request: CreateTodoRequest,
        session: Session = Depends(get_db),
) -> ToDoSchema:
    todo = ToDo.create(request=request) # id None
    todo: ToDo = create_todo(session=session, todo=todo)    # id int
    return ToDoSchema.from_orm(todo)


# 할 일 수정
@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
        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 Found")

# 할 일 삭제
@app.delete("/todos/{todo_id}", status_code=204)
def delete_todo_handler(
        todo_id: int,
        session: Session = Depends(get_db),
):
    todo: ToDo | None = get_todo_by_todo_id(session=session, todo_id=todo_id)

    if not todo:
        raise HTTPException(status_code=404, detail="ToDo Not Found")

    delete_todo(session=session, todo_id=todo_id)

---

💡 Refactoring

database > repository.py

router = APIRouter(prefix="/todos")


@router.get("", status_code=200)
def get_todos_handler(
        order: str | None = None,
        todo_repo: ToDoRepository = Depends(ToDoRepository),
):
    todos: List[ToDo] = todo_repo.get_todos()

    if order == "DESC":
        return ToDoListSchema(
        todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
    )
    return ToDoListSchema(
        todos=[ToDoSchema.from_orm(todo) for todo in todos]
    )

 

Depends 인젝션을 걸어놨기 때문에 request 요청이 들어왔을 때 ToDoRepository 를 주입해주게 된다.

 

api > todo.py

class ToDoRepository:
    def __init__(self, session: Session = Depends(get_db)):
        self.session = session

 

그럼 재귀적으로 호출이 가능하도록 get db 가 의존 주입이 되어있는데.

 

database > connection

def get_db():
    session = SessionFactory()
    try:
        yield session
    finally:
        session.close()

 

get_db 는 session 을 만들고 끝나면 닫아주는 역할을 한다.