FastAPI 섹션 4. 리팩터링

---

1. FastAPI Router

API가 많아지면 main.py 하나에 모든 코드를 넣으면 관리하기 어렵습니다.
Router를 사용하면 API를 기능별로 파일을 분리해서 관리할 수 있습니다.

파일 구조:

src/
├── api/
│   └── todo.py       ← Todo 관련 API를 여기로 분리
├── main.py           ← router만 등록하는 진입점

api/todo.py

from typing import List
from database.connection import get_db
from database.orm import ToDo
from database.repository import get_todos, get_todo_by_todo_id, create_todo, update_todo, delete_todo
from fastapi import Depends, HTTPException, Body, APIRouter
from schema.request import CreateToDoRequest
from schema.response import ToDoListSchema, ToDoSchema
from sqlalchemy.orm import Session

# prefix="/todos" : 이 router의 모든 경로 앞에 /todos가 자동으로 붙음
# 덕분에 각 핸들러에서 "/todos"를 반복해서 쓰지 않아도 됨
router = APIRouter(prefix="/todos")

@router.get("", 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 ToDoListSchema(
            todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
        )
    return ToDoListSchema(
        todos=[ToDoSchema.from_orm(todo) for todo in todos]
    )

@router.get("/{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")

@router.post("", status_code=201)
def create_todo_handler(
    request: CreateToDoRequest,
    session: Session = Depends(get_db),
) -> ToDoSchema:
    todo: ToDo = ToDo.create(request=request)  # id=None (DB 저장 전)
    todo: ToDo = create_todo(session=session, todo=todo)  # id=int (DB 저장 후)
    return ToDoSchema.from_orm(todo)

@router.patch("/{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.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")

@router.delete("/{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)

main.py (router 등록)

from fastapi import FastAPI
from api.todo import router as todo_router

app = FastAPI()

# main.py에서는 router만 등록
app.include_router(todo_router)

Before vs After:

• Before: @app.get("/todos"), @app.get("/todos/{todo_id}") ... 모두 main.py에
• After: @router.get(""), @router.get("/{todo_id}") ... api/todo.py로 분리

---

2. 의존성 주입 (Dependency Injection)

개념 이해

Coupling(결합) : 두 컴포넌트가 강하게 연결되어 있으면 하나를 바꿀 때 다른 것도 바꿔야 합니다.

Dependency(의존성) : 한 컴포넌트가 올바르게 동작하기 위해 다른 요소에 의존하는 것입니다.

Dependency Injection(의존성 주입) : 컴포넌트가 직접 의존성을 만들지 않고 외부에서 주입받는 방식입니다.

예시 코드

class Body:
    def __init__(self, color):
        self.color = color

class Car:
    def __init__(self, body):
        # Car가 직접 Body를 만들지 않고, 외부에서 주입받음
        self.body = body

    def drive(self):
        pass

# 외부에서 Body를 만들어 Car에 주입
red_car = Car(body=Body(color='red'))
blue_car = Car(body=Body(color='blue'))

Car가 Body를 직접 생성하지 않기 때문에 색상이 바뀌어도 Car 코드를 수정할 필요가 없습니다.
FastAPI에서 Depends()가 바로 이 역할을 합니다.

---

3. Repository Pattern (레포지토리 패턴)

개념

Repository Pattern 은 데이터를 다루는 부분을 비즈니스 로직과 분리하는 기술입니다.

• DB 조회/저장 같은 구체적인 구현을 캡슐화하여 숨깁니다.
• 비즈니스 로직(API 핸들러)은 데이터가 어디서 오는지 몰라도 됩니다.
• 나중에 DB를 바꿔도 Repository만 수정하면 됩니다.

Before (함수 방식):

# 함수들이 흩어져 있고, session을 매번 직접 넘겨야 함
todos = get_todos(session=session)
todo = get_todo_by_todo_id(session=session, todo_id=todo_id)

After (Repository 클래스 방식):

# 하나의 클래스로 묶여 있고, session 관리도 내부에서 처리
todos = todo_repo.get_todos()
todo = todo_repo.get_todo_by_todo_id(todo_id=todo_id)

---

database/repository.py

from typing import List
from sqlalchemy import select, delete
from sqlalchemy.orm import Session
from fastapi import Depends
from database.orm import ToDo
from database.connection import get_db

class ToDoRepository:
    def __init__(self, session: Session = Depends(get_db)):
        # FastAPI의 Depends로 세션을 자동 주입받음
        # 직접 세션을 생성하거나 닫을 필요 없음
        self.session = session

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

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

    def create_todo(self, todo: ToDo) -> ToDo:
        self.session.add(instance=todo)
        self.session.commit()           # DB에 저장
        self.session.refresh(instance=todo)  # 저장 후 DB에서 최신 상태 반영 (id 등)
        return todo

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

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

---

api/todo.py (Repository 패턴 적용 후)

from typing import List
from database.orm import ToDo
from database.repository import ToDoRepository
from fastapi import Depends, HTTPException, Body, APIRouter
from schema.request import CreateToDoRequest
from schema.response import ToDoListSchema, ToDoSchema

router = APIRouter(prefix="/todos")

@router.get("", status_code=200)
def get_todos_handler(
    order: str | None = None,
    todo_repo: ToDoRepository = Depends(),  # ToDoRepository 자동 생성 및 주입
) -> ToDoListSchema:
    todos: List[ToDo] = todo_repo.get_todos()
    if order and 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]
    )

@router.get("/{todo_id}", status_code=200)
def get_todo_handler(
    todo_id: int,
    todo_repo: ToDoRepository = Depends(),
) -> ToDoSchema:
    todo: ToDo | None = todo_repo.get_todo_by_todo_id(todo_id=todo_id)
    if todo:
        return ToDoSchema.from_orm(todo)
    raise HTTPException(status_code=404, detail="ToDo Not Found")

@router.post("", status_code=201)
def create_todo_handler(
    request: CreateToDoRequest,
    todo_repo: ToDoRepository = Depends(),
) -> ToDoSchema:
    todo: ToDo = ToDo.create(request=request)       # id=None (DB 저장 전)
    todo: ToDo = todo_repo.create_todo(todo=todo)   # id=int  (DB 저장 후)
    return ToDoSchema.from_orm(todo)

@router.patch("/{todo_id}", status_code=200)
def update_todo_handler(
    todo_id: int,
    is_done: bool = Body(..., embed=True),
    todo_repo: ToDoRepository = Depends(),
):
    todo: ToDo | None = todo_repo.get_todo_by_todo_id(todo_id=todo_id)
    if todo:
        todo.done() if is_done else todo.undone()
        todo: ToDo = todo_repo.update_todo(todo=todo)
        return ToDoSchema.from_orm(todo)
    raise HTTPException(status_code=404, detail="ToDo Not Found")

@router.delete("/{todo_id}", status_code=204)
def delete_todo_handler(
    todo_id: int,
    todo_repo: ToDoRepository = Depends(),
):
    todo: ToDo | None = todo_repo.get_todo_by_todo_id(todo_id=todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="ToDo Not Found")
    todo_repo.delete_todo(todo_id=todo_id)

Depends() 안에 아무것도 넣지 않으면 FastAPI가 타입 힌트(ToDoRepository)를 보고 자동으로 인스턴스를 생성하고 주입해줍니다.

---

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() # 요청이 끝나면 반드시 세션 닫기

---

4. Repository 패턴 적용 후 테스트 코드

기존에는 mocker.patch("main.get_todos", ...) 처럼 함수 경로를 직접 지정했지만,
Repository 패턴 적용 후에는 mocker.patch.object(ToDoRepository, "메서드명", ...) 으로 클래스의 메서드를 mock합니다.

tests/test_main.py

from database.orm import ToDo
from database.repository import ToDoRepository

def test_health_check(client):
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"ping": "pong"}


def test_get_todos(client, mocker):
    # mocker.patch.object(클래스, "메서드명", return_value=가짜데이터)
    # ToDoRepository의 get_todos 메서드를 가짜 데이터로 대체
    mocker.patch.object(ToDoRepository, "get_todos", return_value=[
        ToDo(id=1, contents="FastAPI Section 0", is_done=True),
        ToDo(id=2, contents="FastAPI Section 1", is_done=False),
    ])

    # order=ASC
    response = client.get("/todos")
    assert response.status_code == 200
    assert response.json() == {
        "todos": [
            {"id": 1, "contents": "FastAPI Section 0", "is_done": True},
            {"id": 2, "contents": "FastAPI Section 1", "is_done": False},
        ]
    }

    # order=DESC (순서 반전)
    response = client.get("/todos?order=DESC")
    assert response.status_code == 200
    assert response.json() == {
        "todos": [
            {"id": 2, "contents": "FastAPI Section 1", "is_done": False},
            {"id": 1, "contents": "FastAPI Section 0", "is_done": True},
        ]
    }


def test_get_todo(client, mocker):
    # 200 성공
    mocker.patch.object(
        ToDoRepository,
        "get_todo_by_todo_id",
        return_value=ToDo(id=1, contents="todo", is_done=True),
    )
    response = client.get("/todos/1")
    assert response.status_code == 200
    assert response.json() == {"id": 1, "contents": "todo", "is_done": True}

    # 404 실패 (존재하지 않는 todo)
    mocker.patch.object(
        ToDoRepository,
        "get_todo_by_todo_id",
        return_value=None,
    )
    response = client.get("/todos/1")
    assert response.status_code == 404
    assert response.json() == {"detail": "ToDo Not Found"}


def test_create_todo(client, mocker):
    # ToDo.create가 어떤 값으로 호출됐는지 추적 (실제로 실행은 됨)
    create_spy = mocker.spy(ToDo, "create")

    # DB 저장 함수는 mock으로 대체
    mocker.patch.object(
        ToDoRepository,
        "create_todo",
        return_value=ToDo(id=1, contents="todo", is_done=True),
    )

    body = {"contents": "test", "is_done": False}
    response = client.post("/todos", json=body)

    # DB 저장 전 ToDo.create의 반환값 검증 (이 시점엔 id가 None이어야 함)
    assert create_spy.spy_return.id is None
    assert create_spy.spy_return.contents == "test"
    assert create_spy.spy_return.is_done is False

    # DB 저장 후 응답값 검증 (mock이 반환한 값 기준)
    assert response.status_code == 201
    assert response.json() == {"id": 1, "contents": "todo", "is_done": True}


def test_update_todo(client, mocker):
    # 200 성공
    mocker.patch.object(
        ToDoRepository,
        "get_todo_by_todo_id",
        return_value=ToDo(id=1, contents="todo", is_done=True),
    )

    # ToDo.undone()이 실제로 호출됐는지 검증하기 위해 mock으로 교체
    undone = mocker.patch.object(ToDo, "undone")

    mocker.patch.object(
        ToDoRepository,
        "update_todo",
        return_value=ToDo(id=1, contents="todo", is_done=False),
    )

    response = client.patch("/todos/1", json={"is_done": False})

    # is_done=False이므로 undone()이 정확히 1번 호출됐어야 함
    undone.assert_called_once_with()
    assert response.status_code == 200
    assert response.json() == {"id": 1, "contents": "todo", "is_done": False}

    # 404 실패
    mocker.patch.object(
        ToDoRepository,
        "get_todo_by_todo_id",
        return_value=None,
    )
    response = client.patch("/todos/1", json={"is_done": True})
    assert response.status_code == 404
    assert response.json() == {"detail": "ToDo Not Found"}


def test_delete_todo(client, mocker):
    # 204 성공
    mocker.patch.object(
        ToDoRepository,
        "get_todo_by_todo_id",
        return_value=ToDo(id=1, contents="todo", is_done=True),
    )
    mocker.patch.object(ToDoRepository, "delete_todo", return_value=None)

    response = client.delete("/todos/1")
    assert response.status_code == 204
    # 204 No Content: 응답 본문 없음 → json() 검증 없음

    # 404 실패
    mocker.patch.object(
        ToDoRepository,
        "get_todo_by_todo_id",
        return_value=None,
    )
    response = client.delete("/todos/1")
    assert response.status_code == 404
    assert response.json() == {"detail": "ToDo Not Found"}

---

5. 전체 흐름 정리

클라이언트 요청
     ↓
main.py (router 등록)
     ↓
api/todo.py (APIRouter) ← 기능별 API 분리
     ↓
ToDoRepository (Depends로 자동 주입) ← DB 로직 캡슐화
     ↓
database/orm.py (ORM 모델)
     ↓
database/connection.py (DB 세션)
     ↓
MySQL

---

6. 핵심 개념 요약

개념	설명
APIRouter	API를 기능별 파일로 분리하는 FastAPI 기능
prefix="/todos"	router의 모든 경로 앞에 자동으로 붙는 공통 경로
Depends()	FastAPI가 자동으로 의존성을 생성하고 주입해주는 기능
Repository Pattern	DB 로직을 비즈니스 로직과 분리하는 설계 패턴
mocker.patch.object	클래스의 특정 메서드를 mock으로 교체
session.refresh()	DB 저장 후 자동 생성된 id 등 최신 상태를 객체에 반영