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

5 실습5 회원가입 API 구현

by yororing 2024. 10. 18.

앞 단계 참조 링크:

 

00 개요

  • 회원가입 API 구현하는 실습 진행

01 회원가입 API 로직 구현

1. request.py에 내용 추가

  • 전에 ToDos에 했던 것과 같이 request 모델링을 먼저 해주는 것
  • class SignUpRequest() 생성
# /c/Users/관리자/Desktop/projects/todos/src/schema/request.py 내용

from pydantic import BaseModel

class CreateToDoRequest(BaseModel):
    contents: str
    is_done: bool
    
class SignUpRequest(BaseModel):				# 추가
    username: str					# 추가
    password: str					# 추가

2. api 디렉토리의 user.py 수정

# /c/Users/관리자/Desktop/projects/todos/src/api/user.py 내용
from fastapi import APIRouter
from schema.request import SignUpRequest				# 추가

router = APIRouter(prefix="/users")

@router.post("/sign-up", status_code=201)
def user_sign_up_handler(request: SignUpRequest):       # request: SignUpRequest 추가
    return True
  • request body 처리하는 것
  • "1.request.py에 내용 추가"에서 schema.request에 정의한 SignUpRequest 클래스를 request로 받음

3. service 디렉토리 생성 및 user.py 파일 생성

  • 비밀번호를 암호화하는 과정을 담은 파일을 따로 생성하기
  • service 디렉토리 생성

  • service 디렉토리 안에 user.py 파일 생성

4. service 디렉토리 안에 user.py 작성

  • bcrypt 라이브러리를 사용하여 비밀번호를 해싱하는 기능 구현하기
# /c/Users/관리자/Desktop/projects/todos/src/service/user.py 내용
import bcrypt

class UserService:
    encoding: str = "UTF-8"
    def hash_password(self, plain_password: str) -> str:
        '''
        input:
            - plain_password: string
        output:
            - hashed_password: string
        '''
        hashed_password: bytes = bcrypt.hashpw(
            plain_password.encode(self.encoding),
            salt=bcrypt.gensalt()
        )
        return hashed_password.decode(self.encoding)
  • import bcrypt
    • bcrypt 라이브러리 import 하기 - 비밀번호를 안전하게 해싱하고 검증 시 사용됨
  • class UserService:
        encoding: str = "UTF-8"
    •  UserService 클래스 정의: 비밀번호 해싱과 관련된 메서드 포함, 속성으로 encoding (문자열 "UTF-8")을 갖고 있음
  • def hash_password(self, plain_password: str) -> str:
    • 사용자로부터 받은 평문 비밀번호 (plain_password)를 해싱하여 반환하는 기능을 할 것
    • 입력값으로 str을 받고, 반환값으로 str을 줄 것
  • hashed_password: bytes = bcrypt.hashpw(
                plain_password.encode(self.encoding),
                salt=bcrypt.gensalt()
            )
    • 비밀번호 해싱 과정이 이루어짐
    • plain_password.encode(self.encoding)
      • 사용자가 입력한 평문 비밀번호를 "UTF-8" 형식으로 인코딩하여 byte 형식으로 변환
      • bcrypt는 바이트 형식의 데이터를 사용하기에 인코딩 과정 필수
    • salt = bcrypt.gensalt()
      • 해싱을 할 때 사용될 salt 값 생성.
      • 이 salt는 비밀번호 해싱 시 랜덤한 데이터를 추가하여 동일한 비밀번호가 동일하게 해싱되지 않도록 보
    • bcrypt.hashpw(...)
      • bcrypt.hashpw()가 인코딩된 평문 비밀번호와 생성된 salt를 이용해 해싱된 비밀번호를 생성
      • 결과는 byte 형식의 hashed_password임
  • return hashed_password.decode(self.encoding)
    • 해싱된 비밀번호가 byte 형식으로 저장되었기 때문에 다시  "UTF-8"로 디코딩하여 문자열로 변환한 후 반환함
    • 이 문자열을 DB에 저장하거나 사용자의 비밀번호 검증에 사용할 수 있음

5. api 디렉토리의 user.py 재수정

# /c/Users/관리자/Desktop/projects/todos/src/api/user.py 내용
from fastapi import APIRouter, Depends          # Depends 추가
from schema.request import SignUpRequest
from service.user import UserService            # 추가

router = APIRouter(prefix="/users")

@router.post("/sign-up", status_code=201)
def user_sign_up_handler(
    request: SignUpRequest,                     # 추가
    user_service: UserService = Depends()       # 추가
):
    # 해싱된 비밀번호 생성
    hashed_password: str = user_service.hash_password(
        plain_password = request.password
    )
    return True
  • from fastapi import APIRouter, Depends
    • APIRouter: FastAPI에서 여러 라우터를 관리하고 URL 경로를 그룹화할 때 사용되는 클래스, API 경로를 모듈화하고 구조화 가능
  • from service.user import UserService
    • 우리가 service.user에 작성한 것, 비밀번호 해싱과 같은 사용자 관련 서비스를 처리하는 클래스
  • router = APIRouter(prefix="/users")
    • APIRouter는 FastAPI에서 URL 경로를 그룹화하는 도구로 사용됨
    • prefix="/users"는 이 라우터에서 정의한 모든 경로가 /users로 시작하게 만든다는 의미
    • 예, 이 라우터에서 정의된 /sign-up  경로는 전체 경로로는 /users/sign-up이 됨
  • @router.post("/sign-up", status_code=201)
    def user_sign_up_handler(
        request: SignUpRequest,
        user_service: UserService = Depends()
    ):
    • 이 함수는 /users/sign-up 경로로 POST 요청이 들어왔을 때 호출됨
    • @router.post("/sign-up", status_code=201)
      • FastAPI의 데코레이터
      • 이 함수가 POST 요청을 처리하며 경로는 /sign-up임
      • 회원가입이 성공적으로 처리되면 응답 상태 코드로 201(Created)이 반환됨
    • request: SignUpRequest
      • 이 매개변수는 사용자로부터 요청받은 데이터를 나타냄
      • SignUpRequest 클래스는 이 데이터를 스키마화하여 적절한 형식으로 검증함
      • username, password를 포함 (schema.request.py 모듈 참조)
    • user_service: UserService = Depends()
      • Depends()를 통해 FastAPI의 의존성 주입 기능 사용
      • UserService는 의존성으로 주입되며, 이 서비스는 비밀번호 해싱 같은 사용자 관련 작업을 수행
      • FastAPI는 이 클래스의 인스턴스를 생성하여 user_sign_up_handler 함수의 user_service 매개변수로 전달함
  • hashed_password: str = user_service.hash_password(
        plain_password=request.password
    )
    • user_service.hash_password(plain_password=request.password)
      • 이 부분에서 user_service가 비밀번호 해싱 작업을 담당함
      • 사용자가 요청한 request.password(평문 비밀번호)를 받아서 hash_password 메서드를 호출하여 해싱된 비밀번호를 생성함
      • hashed_password는 해싱된 비밀번호가 저장된 문자열임
  • return True
    • 회원가입이 성공적으로 처리된 경우 True 값 반환

 

6. orm.py 수정

# /c/Users/관리자/Desktop/projects/todos/src/database/orm.py 내용
from sqlalchemy import Boolean, Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base, relationship
from schema.request import CreateToDoRequest

Base = declarative_base()

...

# User 클래스 모델링 한 것
class User(Base):
    __tablename__ = 'user'
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(256), unique=True, nullable=False)
    password = Column(String(256), nullable=False)
    todos = relationship("ToDo", lazy="joined")
    
    @classmethod							# 추가 (이 줄부터 끝까지)
    def create(cls, username: str, hashed_password: str) -> "User":
        return cls(
            username = username,
            password=hashed_password,
        )
  • Base = declarative_base()
    • declarative_base()는 sqlalchemy에서 사용되는 기본 클래스, 이를 통해 테이블과 Python 클래스 매핑함
    • Base는 모든 DB 모델 클래스가 상속받아햐 하는 기반 클래스로 사용됨
  • class User(Base):
        __tablename__ = 'user'
    • User는 Base 클래스를 상속받아 sqlalchemy에서 사용할 수 있는 모델로 만듬
    • 이 클래스는 DB의 'user' 테이블과 매핑됨, 즉 User 클래스의 인스턴스는 user 테이블의 행(row)과 대응
  • @classmethod
    def create(cls, username: str, hashed_password: str) -> "User":
        return cls(
            username=username,
            password=hashed_password,
        )
    • @classmethod
      • 클래스 메서드로 정의되었으며, 인스턴스가 아닌 클래스 자체에서 호출 가능
      • 메서드의 첫 번째 매개변수로 클래스 자체를 받는 cls를 사용함
    • def create(cls, username: str, hashed_password: str)
      • 이 메서드는 사용자의 username과 해싱된 비밀번호(hashed_password)를 매개변수로 받아 새로운 User 객체를 생성
    • return cls(username=username, password=hashed_password)
      • 주어진 username과 해싱된 비밀번호를 사용해 새로운 User 객체를 반환
      • 반환된 객체는 DB에 저장되기 전 준비된 상태

 

7. api 디렉토리의 user.py 재수정

# /c/Users/관리자/Desktop/projects/todos/src/api/user.py 내용
from fastapi import APIRouter, Depends
from schema.request import SignUpRequest
from service.user import UserService
from database.orm import User                   # 추가

router = APIRouter(prefix="/users")

@router.post("/sign-up", status_code=201)
def user_sign_up_handler(
    request: SignUpRequest,
    user_service: UserService = Depends()
):
    # 해싱된 비밀번호 생성
    hashed_password: str = user_service.hash_password(
        plain_password = request.password
    )
    # 사용자 생성                               # 추가 (여기서 아래까지)
    user: User = User.create(
        username = request.username,
        hashed_password = hashed_password
    )
    
    return True
  • repository 패턴을 사용하여 여기서 생성된 사용자를 DB에 저장할 것

8. 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 fastapi import Depends

from database.orm import ToDo, User             # User 추가
from database.connection import get_db

...

class UserRepository:                       # 추가 (여기부터 아래까지)
    def __init__(self, session: Session = Depends(get_db)):
        self.session = session
    
    def save_user(self, user: User) -> User:
        self.session.add(instance=user)
        self.session.commit()
        self.session.refresh(instance=user)
        return user

9. api의 user.py 재수정

# /c/Users/관리자/Desktop/projects/todos/src/api/user.py 내용
from fastapi import APIRouter, Depends
from schema.request import SignUpRequest
from service.user import UserService
from database.orm import User
from database.repository import UserRepository  # 추가

router = APIRouter(prefix="/users")

@router.post("/sign-up", status_code=201)
def user_sign_up_handler(
    request: SignUpRequest,
    user_service: UserService = Depends(),
    user_repo: UserRepository = Depends()       # 추가
):
    # 해싱된 비밀번호 생성
    hashed_password: str = user_service.hash_password(
        plain_password = request.password
    )
    # 사용자 생성
    user: User = User.create(
        username = request.username,
        hashed_password = hashed_password
    )
    # 사용자를 DB에 저장                         # 추가 (여기부터 아래까지)
    user: User = user_repo.save_user(user=user) # 이 시점에서 사용자의 id 값 = int로 들어감
    
    return True
  • 이제 user를 반환할 것인데, 그냥 반환하지 않고 마찬가지로 response를 정의하여 반환할 것

10. 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
        orm_mode = True # 이거 추가 해야지 ConfigError 안남

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

class UserSchema(BaseModel):				# 추가 (여기부터 아래까지)
    id: int
    username: str
    
    class Config:
        from_attributes = True
        orm_mode = True # 이거 추가 해야지 ConfigError 안남
  • id 값과 username 값 반환
  • password가 해싱되어있긴 하지만 굳이 그 해시 값을 노출할 필요가 없기에 password는 반환하지 않음 
  • orm_mode = True를 포함한 class Config를 추가 해줘야지 error가 안나기에 추가

 

11. api의 user.py 재수정

  • response.py에 정의된 UserSchema를 반환값으로 설정
# /c/Users/관리자/Desktop/projects/todos/src/api/user.py 내용
from fastapi import APIRouter, Depends
from schema.request import SignUpRequest
from schema.response import UserSchema          # 추가
from service.user import UserService
from database.orm import User
from database.repository import UserRepository

router = APIRouter(prefix="/users")

@router.post("/sign-up", status_code=201)
def user_sign_up_handler(
    request: SignUpRequest,
    user_service: UserService = Depends(),
    user_repo: UserRepository = Depends()
):
    # 해싱된 비밀번호 생성
    hashed_password: str = user_service.hash_password(
        plain_password = request.password
    )
    # 사용자 생성
    user: User = User.create(
        username = request.username,
        hashed_password = hashed_password
    )
    # 사용자를 DB에 저장
    user: User = user_repo.save_user(user=user) # 이 시점에서 사용자의 id 값 = int로 들어감

    # user(id, username) 값 반환                # 추가
    return UserSchema.from_orm(user)
  • request: SignUpRequest
    • 우리가 정의한 pydantic 클래스를 통해 request로 SignUpRequest를 전달 받는 것
    • 이 request에는 username과 password가 있음
  • hashed_password: str = user_service.hash_password( plain_password = request.password )
    •  request.password를 암호화 하는 과정
  • user: User = User.create( username = request.username, hashed_password = hashed_password )
    • request.username과 바로 위에서 생성한 hashed_password로  user 객체 생성하는 과정 
  • user: User = user_repo.save_user(user=user)
    • 바로 위에서 만든 user를 DB에 저장하는 과정
  • return UserSchema.from_orm(user)
    • 저장된 user 값을 UserSchema.from_orm을 통해 refresh해서 response로 응답하는 구조임!

 

이때까지의 코드들: 13개

  • /c/Users/관리자/Desktop/projects/todos/src/main.py
  • /c/Users/관리자/Desktop/projects/todos/src/api/todo.py
  • /c/Users/관리자/Desktop/projects/todos/src/api/user.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/service/user.py
  • /c/Users/관리자/Desktop/projects/todos/src/tests/test_main.py
  • /c/Users/관리자/Desktop/projects/todos/src/tests/test_todos_api.py
  • /c/Users/관리자/Desktop/projects/todos/src/tests/test_users_api.py
  • /c/Users/관리자/Desktop/projects/todos/src/tests/conftest.py

 

이때까지의 디렉토리

TODOS

├── Include

├── Lib

├── Scripts

├── src

│     ├── api

│     │      ├── todo.py

│     │      └── user.py

│     ├── database

│     │      ├── connection.py

│     │      ├── orm.py

│     │      └── repository.py

│     ├── schema

│     │      ├── request.py

│     │      └── response.py

│     ├── service

│     │      └── user.py

│     ├── tests

│     │      ├── __init__.py

│     │      ├── conftest.py

│     │      ├── test_main.py

│     │      ├── test_todos_api.py

│     │      └── test_users_api.py

│      └──  main.py

└── pyvenv.cfg

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

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

# 첫 화면 API
@app.get("/")
def health_check_handler():
    return {"ping": "pong"}
# /c/Users/관리자/Desktop/projects/todos/src/api/todo.py 내용
from fastapi import Body, HTTPException, Depends, APIRouter
from typing import List	

from database.repository import ToDoRepository, ToDo
from schema.response import ToDoSchema, ToDoListSchema
from 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)
  • api/user.py 수정
# /c/Users/관리자/Desktop/projects/todos/src/api/user.py 내용
from fastapi import APIRouter, Depends
from schema.request import SignUpRequest
from schema.response import UserSchema          # 추가
from service.user import UserService
from database.orm import User
from database.repository import UserRepository

router = APIRouter(prefix="/users")

@router.post("/sign-up", status_code=201)
def user_sign_up_handler(
    request: SignUpRequest,
    user_service: UserService = Depends(),
    user_repo: UserRepository = Depends()
):
    # 해싱된 비밀번호 생성
    hashed_password: str = user_service.hash_password(
        plain_password = request.password
    )
    # 사용자 생성
    user: User = User.create(
        username = request.username,
        hashed_password = hashed_password
    )
    # 사용자를 DB에 저장
    user: User = user_repo.save_user(user=user) # 이 시점에서 사용자의 id 값 = int로 들어감

    # user(id, username) 값 반환                # 추가
    return UserSchema.from_orm(user)
# /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()
  • orm.py 수정
# /c/Users/관리자/Desktop/projects/todos/src/database/orm.py 내용
from sqlalchemy import Boolean, Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base, relationship
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)
    user_id = Column(Integer, ForeignKey("user.id"))
    
    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

# User 클래스 모델링 한 것
class User(Base):
    __tablename__ = 'user'
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(256), unique=True, nullable=False) # unique=True는 내가 넣은 것임
    password = Column(String(256), nullable=False)
    todos = relationship("ToDo", lazy="joined")
    
    @classmethod
    def create(cls, username: str, hashed_password: str) -> "User":
        return cls(
            username = username,
            password=hashed_password,
        )
  • 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 fastapi import Depends

from database.orm import ToDo, User
from 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.session.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()

class UserRepository:
    def __init__(self, session: Session = Depends(get_db)):
        self.session = session
    
    def save_user(self, user: User) -> User:
        self.session.add(instance=user)
        self.session.commit()
        self.session.refresh(instance=user)
        return user
  • request.py 수정
# /c/Users/관리자/Desktop/projects/todos/src/schema/request.py 내용

from pydantic import BaseModel

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

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

class UserSchema(BaseModel):
    id: int
    username: str
    
    class Config:
        from_attributes = True
        orm_mode = True # 이거 추가 해야지 ConfigError 안남
  • service/user.py 추가
# /c/Users/관리자/Desktop/projects/todos/src/service/user.py 내용
import bcrypt

class UserService:
    encoding: str = "UTF-8"
    def hash_password(self, plain_password: str) -> str:
        '''
        input:
            - plain_password: string
        output:
            - hashed_password: string
        '''
        hashed_password: bytes = bcrypt.hashpw(
            plain_password.encode(self.encoding),
            salt=bcrypt.gensalt()
        )
        return hashed_password.decode(self.encoding)
# /c/Users/관리자/Desktop/projects/todos/src/tests/conftest.py 내용
import pytest
from fastapi.testclient import TestClient
from main import app

@pytest.fixture
def client():
	return TestClient(app=app)
# /c/Users/관리자/Desktop/projects/todos/src/tests/test_main.py 내용

# GET Method 사용하여 "/" 검증
def test_health_check(client):      # client 는 내가 conftest.py에서 정의한 fixture
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"ping": "pong"}
# /c/Users/관리자/Desktop/projects/todos/src/tests/test_todos_api.py 내용
from src.schema.response import ToDoSchema
from src.database.orm import ToDo
from src.database.repository import ToDoRepository

# 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/test_users_api.py 내용
def test_user_sign_up(client):
    response = client.post("users/sign-up")
    assert response.status_code == 200
    assert response.json() is True

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

5 기능 고도화 - 로그인 / 유저 인증 - JWT  (0) 2024.11.16
5 실습6 회원가입 API 테스트  (2) 2024.11.16
5 실습4 회원가입 API 생성 & 비밀번호 암호화(bcrypt)  (1) 2024.10.18
5 기능 고도화 Lazy Loading / Eager Loading  (0) 2024.10.18
5 실습3 ORM JOIN  (1) 2024.10.16

관련글

티스토리툴바