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

4 실습3 Refactoring - Repository Pattern 레포지토리 패

by yororing 2024. 8. 8.

앞 단계 참조 링크:

 

00 개요

  • 함수로 구현된 reposistory pattern을 class로 정의(모듈화)해주는 실습 진행 

01 Repository Pattern이란

1. 정의 및 개념

  • 데이터를 다루는 부분을 추상화하는 기술로 비즈니스 로직과 데이터 관리의 강한 결합을 없애준다
  • 데이터를 불러오고 저장하는 것과 같은 구체적인 구현은 감춘다
  • 우리 프로젝트에선 이미 repository pattern이 구현되어 있음

2. 예시

  • todo.py에 보면,
# GET Method 사용하여 전체 조회 API
@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":				# get_todos() 함수를 실행시켜 불러옴
        return ToDoListSchema(
            todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
        )
    return ToDoListSchema(
        todos=[ToDoSchema.from_orm(todo) for todo in todos]
    )
  • 어떤 데이터베이스를 다루는 코드를 적어준 것이 아니라, get_todos()라는 함수를 그저 호출해서 사용하는 형태로 코드가 구현되어있음
  • get_todos() 함수의 내용을 보면,

  • return list(session.scalars(select(ToDo)))와 같이, ToDo 안에서 session에서 어떠한 구체적인 방법을 통해 데이터를 조회하는 것을 구현해 놨으며
  • get_todos() 아닌 외부에서는 필요한 인자만 전달하여 결과값만 받는 형태로 구현되어 있음 
  • 이런식으로 데이터를 처리하는 기술을 repository pattern이라고 부름
  • 우리가 생성한 프로젝트에서는 함수의 형태로 repository pattern이 구현되어 있음
# /c/Users/관리자/Desktop/projects/todos/src/database/repository.py 내용
from sqlalchemy import select, delete
from sqlalchemy.orm import Session
from typing import List

from src.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

def delete_todo(session: Session, todo_id: ToDo) -> None:
    session.execute(delete(ToDo).where(ToDo.id == todo_id))
    session.commit()
  • 함수마다 인자를 보면 동일한 session이라는 인자를 전달 받고 있음
  • 이럴 경우 class로 모듈화 하면 코드의 가독성이 향상되고 코드의 응집력이 향상됨
  • 그래서 해당 repository pattern을 class로 바꾸는 실습 진행할 것!

02 class 생성 및 적용

1. repository.py 수정하기 - class 생성

  • 원래 repository.py 내용:
# /c/Users/관리자/Desktop/projects/todos/src/database/repository.py 내용
from sqlalchemy import select, delete
from sqlalchemy.orm import Session
from typing import List

from src.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

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

1) repository.py 맨 윗부분에 클래스 정의하고, 작성되어 있었던 함수들은 한 번 씩 tab 해줘서 class ToDoRepository() 에 추가해주기 

  • 다음을 repository.py 맨 윗부분에 추가:
# /c/Users/관리자/Desktop/projects/todos/src/database/repository.py 내용
from fastapi import Depends						# 추가
from src.database.connection import get_db				# 추가

class ToDoRepository:
    def __init__(self, session:Session = Depends(get_db)):    # init 메소드 overwrite 해주기
    self.session = session                                   # session을 인자로 받아서 사용하도록 함

2) 기존 함수들에 있었던 인자 (session:Session) 부분 수정해주기

  • def 함수이름(session:Session) → def 함수이름(self) 로 변경
  • 함수 내에서 사용된 session → self.session으로 변경
# /c/Users/관리자/Desktop/projects/todos/src/database/repository.py 내용
from sqlalchemy import select, delete
from sqlalchemy.orm import Session
from typing import List
from fastapi import Depends

from src.database.orm import ToDo
from src.database.connection import get_db

class ToDoRepository:
    def __init__(self, session:Session = Depends(get_db)):
        self.session = session 
        
    def get_todos(self,) -> List[ToDo]:				# session: Session -> self로 변경
        return list(self.session.scalars(select(ToDo)))         # session -> self.session으로 변경

    def get_todo_by_todo_id(self, todo_id: int) -> ToDo | None:        # session: Session -> self로 변경
        return self.session.scalar(select(ToDo).where(ToDo.id == todo_id))  # session -> self.session으로 변경

    def create_todo(self, todo: ToDo) -> ToDo:        # session: Session -> self로 변경
        self.session.add(instance=todo)              # session -> self.session으로 변경
        self.session.commit()                        # session -> self.session으로 변경
        self.session.refresh(instance=todo)          # session -> self.session으로 변경
        return todo

    def update_todo(self, todo: ToDo) -> ToDo:         # session: Session -> self로 변경
        self.session.add(instance=todo)                # session -> self.session으로 변경
        self.session.commit()                          # session -> self.session으로 변경
        self.session.refresh(instance=todo)             # session -> self.session으로 변경
        return todo

    def delete_todo(self, todo_id: ToDo) -> None:          # session: Session -> self로 변경
        self.session.execute(delete(ToDo).where(ToDo.id == todo_id))       # session -> self.session으로 변경
        self.session.commit()                             # session -> self.session으로 변경

 

2. todo.py (API 파일) 수정하기

  • 원래 repository.py 내용:
# /c/Users/관리자/Desktop/projects/todos/src/api/todo.py 내용
from fastapi import FastAPI, Body, HTTPException, Depends, APIRouter
from sqlalchemy.orm import Session
from typing import List	

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

router = APIRouter(prefix='/todos')


# GET Method 사용하여 전체 조회 API
@router.get("", status_code=200)            # "" 안에 /todos 제거됨
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]
    )

# GET Method 사용하여 단일 조회 API
@router.get("/{todo_id}", status_code=200)            # "" 안에 /todos 제거됨
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")

# POST Medthod 사용하여 todo 생성 API
@router.post("", status_code=201)            # "" 안에 /todos 제거됨
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
@router.patch("/{todo_id}", status_code=200)            # "" 안에 /todos 제거됨
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 Found")

# DELETE Method 사용하여 todo 아이템 삭제 API
@router.delete("/{todo_id}", status_code=204)            # "" 안에 /todos 제거됨
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)
  • import 부분에서
    • 'from src.database.repository import 함수들'에서 '함수들' 를 ToDoRepository 로 변경
  • 인자로 session을 받았던 부분 수정
    • 'session: Session = Depends(get_db),' → 'todo_repo: ToDoRepository = Depends(ToDoRepository),'
    • 'todos: List[ToDo] = get_todos(session=session)' → 'todos: List[ToDo] = todo_repo.get_todos()'
# /c/Users/관리자/Desktop/projects/todos/src/api/todo.py 내용
from fastapi import Body, HTTPException, Depends, APIRouter
from typing import List	

from src.database.repository import ToDoRepository, ToDo 		# 수정
from src.schema.response import ToDoSchema, ToDoListSchema
from src.schema.request import CreateToDoRequest

router = APIRouter(prefix='/todos')


# GET Method 사용하여 전체 조회 API
@router.get("", status_code=200)            # "" 안에 /todos 제거됨
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]
    )

# GET Method 사용하여 단일 조회 API
@router.get("/{todo_id}", status_code=200)            # "" 안에 /todos 제거됨
def get_todo_handler(
    todo_id: int,
    todo_repo: ToDoRepository = Depends(ToDoRepository), 		     # 수정
    ) -> 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")

# POST Medthod 사용하여 todo 생성 API
@router.post("", status_code=201)            # "" 안에 /todos 제거됨
def create_todo_handler(
    request: CreateToDoRequest,
    todo_repo: ToDoRepository = Depends(ToDoRepository), 		# 수정
) -> ToDoSchema:
    todo: ToDo = ToDo.create(request=request)  # id=None
    todo: ToDo = todo_repo.create_todo(todo=todo) # id=int 		# 수정
    return ToDoSchema.from_orm(todo)

# PATCH Method 사용하여 is_done 값 수정 API
@router.patch("/{todo_id}", status_code=200)            # "" 안에 /todos 제거됨
def update_todo_handler(
    todo_id: int,
    is_done: bool = Body(..., embed=True),
    todo_repo: ToDoRepository = Depends(ToDoRepository), 		    # 수정
    ):
    todo:ToDo | None = todo_repo.get_todo_by_todo_id(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 = todo_repo.update_todo(todo=todo) 		# 수정
        return ToDoSchema.from_orm(todo)
    raise HTTPException(status_code=404, detail="ToDo Not Found")

# DELETE Method 사용하여 todo 아이템 삭제 API
@router.delete("/{todo_id}", status_code=204)               # "" 안에 /todos 제거됨
def delete_todo_handler(
    todo_id: int,
    todo_repo: ToDoRepository = Depends(ToDoRepository), 		        # 수정
    ):
    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)
  • 설명:
    • todo.py에 보면,
더보기

# /c/Users/관리자/Desktop/projects/todos/src/api/todo.py 내용:

# GET Method 사용하여 전체 조회 API
@router.get("", status_code=200)            # "" 안에 /todos 제거됨
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]
    )

  • 6번째 줄:
    • Dependency Injection을 걸어놨기 때문에 FastAPI가 request 요청이 들어왔을 때 이 Depends() 안에 있는 ToDoRepository를 주입함
    • 해당 ToDoRepository가 정의된 repository.py에 가보면,
더보기

# /c/Users/관리자/Desktop/projects/todos/src/database/repository.py 내용:

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

  • 3번째 줄
    • get_db가 Dependency Injection으로 적용이 되어 있음
    • 그러므로 FastAPI가 이 Dependency Injection을 recursive call (i.e., 재귀 호출)을 해서 여기 걸려 있는 의존성 역시 호출해주게 됨
더보기

1    # /c/Users/관리자/Desktop/projects/todos/src/database/connection.py 내용
2    from sqlalchemy import create_engine
3    from sqlalchemy.orm import sessionmaker
4 
5    DATABASE_URL = "mysql+pymysql://root:todos@127.0.0.1:3306/todos"
6 
7    engine = create_engine(DATABASE_URL, echo=True)
8    SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)
9
10    def get_db():
11       session = SessionFactory()
12       try:
13           yield session
14       finally:
15           session.close()

  • 10번째 줄
    • get_db는 connection.py에 있는 함수
    • request 요청이 들어올 때 get_db를 실행
    • get_db는 SessionFactory()를 사용하여 session을 생성해 준 다음
  • 13번째 줄
    • 맨 마지막의 request가 처리되고
  • 15번째 줄
    • session을 close해줌 
  • 이런식으로, FastAPI의 Dependency Injection 같은 경우는 계속 재귀적 호출이 가능한 구조로 되어 있음

3. test_main.py (테스트 코드) 수정하기

  • 기존의 repository.py에서 함수들을 class로 수정하였기에 이에 알맞게 test_main.py 도 수정 필요
  • 더 이상 함수를 직접 mocking하지 않고 mocker.patch.object()를 사용하여 ToDoRepository 클래스를 mocking하도록  변경 (import 도 해주기!)
  • 이전에 update 때 object를 통해 mocking 했던 것이 있는데, 그와 동일하게 ToDoRepository 클래스를 mocking하도록 하는 것 
  • 그래서 그 후에 오는 경로들을 지워줘도 됨 
# /c/Users/관리자/Desktop/projects/todos/src/tests/test_main.py 내용
from src.schema.response import ToDoSchema
from src.database.orm import ToDo
from src.database.repository import ToDoRepository                          # 추가

# GET Method 사용하여 "/" 검증
def test_health_check(client):
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"ping": "pong"}

# GET Method 사용하여 전체 조회 API 검증
def test_get_todos(client, mocker):
    # order = ASC
    mocker.patch.object(ToDoRepository, "get_todos", return_value=[          # 수정
        ToDo(id=1, contents="FastAPI Section 0", is_done=True),
        ToDoSchema(id=2, contents="FastAPI Section 2", is_done=False),
    ])
    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 2", "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 2", "is_done": False},
            {"id":1, "contents": "FastAPI Section 0", "is_done": True},
        ]
    }

# GET Method 사용하여 단일 조회 API 검증
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
    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"}

# POST Medthod 사용하여 todo 생성 API 검증
def test_create_todo(client, mocker):
    create_spy = mocker.spy(ToDo, "create")
    mocker.patch.object(ToDoRepository, "create_todo",                 # 수정
        return_value=ToDo(id=1, contents="todo", is_done=False),
    )
    
    body = {
        "contents": "test",
        "is_done": False,
    }
    response = client.post("/todos", json=body)
    
    assert create_spy.spy_return.id is None
    assert create_spy.spy_return.contents == "test"
    assert create_spy.spy_return.is_done is False
    
    assert response.status_code == 201
    assert response.json() == {"id":1, "contents":"todo", "is_done":False}


# PATCH Method 사용하여 is_done 값 수정 API 검증
def test_update_todo(client, mocker):
    # 상태코드 200
    
    # get_todo_by_todo_id 검증
    mocker.patch.object(ToDoRepository, "get_todo_by_todo_id",      # 수정
        return_value = ToDo(id=1, contents="todo", is_done=True),
    )
    
    # done() 또는 undone() 검증
    # done = mocker.patch.object(ToDo, "done") # ToDo 객체의 done() 메소드 호출
    undone = mocker.patch.object(ToDo, "undone") # ToDo 객체의 undone() 메소드 호출
    
    # update_todo 검증
    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})
    
    # done() 또는 undone() 검증 - done 또는 undone이 한 번 호출되었는지 확인. 아닐 경우 AssertionError 발생
    # done.assert_called_once_with()
    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"}
  
# DELTE Method API 검증  
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 
    
    # 상태코드 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"}

 

1) pytest 실행

  • src 디렉토리로 이동하여 $ pytest 실행해보기

  • 정상 작동 확인!

03 Additional Knowledge - Syntax 더 깔끔하게 하기

  • todo.py에 보면, get_todos_handler() 정의 시 인자로 받는 todo_repo: ToDoRepository = Depends(ToDoRepository)에 보면
    • typehint인 ToDoRepository와 Dependency Injection이 적용되는 ToDoRepository 가 같은 경우, Dependency 안에 있는 인자 삭제해줘도 됨!
    • 즉, todo_repo: ToDoRepository = Depends(), 라고 써도 됨!
  • 그리하여 수정된 todo.py 내용:
# /c/Users/관리자/Desktop/projects/todos/src/api/todo.py 내용
from fastapi import Body, HTTPException, Depends, APIRouter
from typing import List	

from src.database.repository import ToDoRepository, ToDo
from src.schema.response import ToDoSchema, ToDoListSchema
from src.schema.request import CreateToDoRequest

router = APIRouter(prefix='/todos')


# GET Method 사용하여 전체 조회 API
@router.get("", status_code=200)
def get_todos_handler(
    order: str | None = None,
    todo_repo: ToDoRepository = Depends(),       # 수정
) -> 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]
    )

# GET Method 사용하여 단일 조회 API
@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")

# POST Medthod 사용하여 todo 생성 API
@router.post("", status_code=201)
def create_todo_handler(
    request: CreateToDoRequest,
    todo_repo: ToDoRepository = Depends(),       # 수정
) -> ToDoSchema:
    todo: ToDo = ToDo.create(request=request)      # id=None
    todo: ToDo = todo_repo.create_todo(todo=todo) # id=int
    return ToDoSchema.from_orm(todo)

# PATCH Method 사용하여 is_done 값 수정 API
@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:
        # update - is_done값이 True이면 todo.done() 실행, False이면 todo.undone() 실행
        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")

# DELETE Method 사용하여 todo 아이템 삭제 API
@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)

 

1) 수정 후 pytest 실행

  • src 디렉토리로 이동하여 $ pytest 실행해보기

  • 정상 작동 확인!

 

 

 

이때까지의 코드들: 9개

  • /c/Users/관리자/Desktop/projects/todos/src/main.py
  • /c/Users/관리자/Desktop/projects/todos/src/api/todo.py - (수정 - 변경)
  • /c/Users/관리자/Desktop/projects/todos/src/database/orm.py
  • /c/Users/관리자/Desktop/projects/todos/src/database/connection.py
  • /c/Users/관리자/Desktop/projects/todos/src/database/repository.py
  • /c/Users/관리자/Desktop/projects/todos/src/schema/request.py
  • /c/Users/관리자/Desktop/projects/todos/src/schema/response.py
  • /c/Users/관리자/Desktop/projects/todos/src/tests/test_main.py (수정 - 변경)
  • /c/Users/관리자/Desktop/projects/todos/src/tests/conftest.py

 

# /c/Users/관리자/Desktop/projects/todos/src/main.py 내용
from fastapi import FastAPI
from src.api import todo

app = FastAPI()
app.include_router(todo.router)

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

 

 

  • todo.py 수정
# /c/Users/관리자/Desktop/projects/todos/src/api/todo.py 내용
from fastapi import Body, HTTPException, Depends, APIRouter
from typing import List	

from src.database.repository import ToDoRepository, ToDo
from src.schema.response import ToDoSchema, ToDoListSchema
from src.schema.request import CreateToDoRequest

router = APIRouter(prefix='/todos')


# GET Method 사용하여 전체 조회 API
@router.get("", status_code=200)
def get_todos_handler(
    order: str | None = None,
    todo_repo: ToDoRepository = Depends(),
) -> 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]
    )

# GET Method 사용하여 단일 조회 API
@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")

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

# PATCH Method 사용하여 is_done 값 수정 API
@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:
        # update - is_done값이 True이면 todo.done() 실행, False이면 todo.undone() 실행
        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")

# DELETE Method 사용하여 todo 아이템 삭제 API
@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)
# /c/Users/관리자/Desktop/projects/todos/src/database/orm.py 내용
from sqlalchemy import Boolean, Column, Integer, String
from sqlalchemy.orm import declarative_base
from src.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 내용
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 내용
from sqlalchemy import select, delete
from sqlalchemy.orm import Session
from typing import List
from fastapi import Depends

from src.database.orm import ToDo
from src.database.connection import get_db

class ToDoRepository:
    def __init__(self, session:Session = Depends(get_db)):
        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.ession.scalar(select(ToDo).where(ToDo.id == todo_id))

    def create_todo(self, todo: ToDo) -> ToDo:
        self.session.add(instance=todo)
        self.session.commit()
        self.session.refresh(instance=todo)
        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: ToDo) -> None:
        self.session.execute(delete(ToDo).where(ToDo.id == todo_id))
        self.session.commit()
# /c/Users/관리자/Desktop/projects/todos/src/schema/request.py 내용

from pydantic import BaseModel

class CreateToDoRequest(BaseModel):
    contents: str
    is_done: bool
# /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
        orm_mode = True # 이거 추가 해야지 ConfigError 안남

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

 

 

  • test_main.py 수정
# /c/Users/관리자/Desktop/projects/todos/src/tests/test_main.py 내용
from src.schema.response import ToDoSchema
from src.database.orm import ToDo
from src.database.repository import ToDoRepository      # 추가

# GET Method 사용하여 "/" 검증
def test_health_check(client):      # client 는 내가 conftest.py에서 정의한 fixture
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"ping": "pong"}

# GET Method 사용하여 전체 조회 API 검증
def test_get_todos(client, mocker):
    # order = ASC
    mocker.patch.object(ToDoRepository, "get_todos", return_value=[          # 수정
        ToDo(id=1, contents="FastAPI Section 0", is_done=True),
        ToDoSchema(id=2, contents="FastAPI Section 2", is_done=False),
    ])
    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 2", "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 2", "is_done": False},
            {"id":1, "contents": "FastAPI Section 0", "is_done": True},
        ]
    }

# GET Method 사용하여 단일 조회 API 검증
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")   # path에 하위 서브 path 적어주기
    assert response.status_code == 200
    assert response.json() == {"id":1, "contents": "todo", "is_done": True }
    
    # 상태코드 404
    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"}

# POST Medthod 사용하여 todo 생성 API 검증
def test_create_todo(client, mocker):
    create_spy = mocker.spy(ToDo, "create")     # mocker의 spy 기능 사용
    mocker.patch.object(ToDoRepository, "create_todo",              # 수정
        return_value=ToDo(id=1, contents="todo", is_done=False),
    )
    
    body = {
        "contents": "test",
        "is_done": False,
    }
    response = client.post("/todos", json=body)
    
    assert create_spy.spy_return.id is None
    assert create_spy.spy_return.contents == "test"
    assert create_spy.spy_return.is_done is False
    
    assert response.status_code == 201
    assert response.json() == {"id":1, "contents":"todo", "is_done":False}


# PATCH Method 사용하여 is_done 값 수정 API 검증
def test_update_todo(client, mocker):
    # 상태코드 200
    
    # get_todo_by_todo_id 검증
    mocker.patch.object(ToDoRepository, "get_todo_by_todo_id",      # 수정
        return_value = ToDo(id=1, contents="todo", is_done=True),
    )
    
    # done() 또는 undone() 검증
    # done = mocker.patch.object(ToDo, "done") # ToDo 객체의 done() 메소드 호출
    undone = mocker.patch.object(ToDo, "undone") # ToDo 객체의 undone() 메소드 호출
    
    # update_todo 검증
    mocker.patch.object(ToDoRepository, "update_todo",              # 수정
        return_value = ToDo(id=1, contents="todo", is_done=False),   # done 경우 is_done = True, undone 경우 is_done=False 주기
    )
    
    response = client.patch("/todos/1", json={"is_done": False})   # done 경우 "is_done": True, undone 경우 "is_done": False 주기
    
    # done() 또는 undone() 검증 - done 또는 undone이 한 번 호출되었는지 확인. 아닐 경우 AssertionError 발생
    # done.assert_called_once_with()
    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"}
  
# DELTE Method API 검증  
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")    # path에 하위 서브 path 적어주기
    assert response.status_code == 204      # status_code만 주고 response.json()는 assert 안 해봐도 됨! 반환값이 없을 것이기에
    
    # 상태코드 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"}
# /c/Users/관리자/Desktop/projects/todos/src/tests/conftest.py 내용
import pytest
from fastapi.testclient import TestClient
from src.main import app

@pytest.fixture
def client():
	return TestClient(app=app)

'Web 개발 > FAST API (인프런 강의 내용)' 카테고리의 다른 글

5 실습1 기능 고도화 User 모델링  (0) 2024.08.12
5 기능 고도화 SQL JOIN  (0) 2024.08.11
4 실습2 Refactoring - Dependency Injection 의존성 주입  (0) 2024.08.08
4 실습1 Refactoring - FastAPI Router  (0) 2024.08.08
3 실습8 테스트 코드 - DELETE API  (0) 2024.08.07

관련글

티스토리툴바