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

5 실습4 회원가입 API 생성 & 비밀번호 암호화(bcrypt)

by yororing 2024. 10. 18.

앞 단계 참조 링크:

 

00 개요

  • 회원가입 API 틀 작성 및 검증하는 실습 진행
  • bcrypt 간단히 사용하여 비밀번호 암호화 (해싱) 하는 실습 진행 

01 회원가입 API 개발

1. user.py 생성

  • todos/src/api 디렉토리 안에 user.py라는 새 파일 생성하기

2. user.py에 API 작성 (일단 간단히)

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

router = APIRouter(prefix="/users")

@router.post("/sign-up")
def user_sign_up_handler():
    return True
  • fastapi에서 APIRouter import 하기
  • APIRouter() 사용하여 router 객체 생성하기, prefix 키워드에 "/users" 매핑해주기
  • router에 post 메소드로 연결하여 API 생성하기, path에 "/sign-up" 설정   
  • user_sign_up_handler() 함수 정의
    • 일단 True 반환값으로 설정

3. main.py에 작성

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

app = FastAPI()
app.include_router(todo.router)
app.include_router(user.router)				# 추가

# 첫 화면 API
@app.get("/")
def health_check_handler():
    return {"ping": "pong"}
  • src.api에서 user import하기
  • app.include_router()에  방금 생성한 user.router (i.e., user.py에 있는 router 객체)를 인자로 넣어주기 → user router 연결

4. 테스트 코드로 API 연동 확인

  • API가 잘 연동 되어있는지 확인하기 위해 테스트 코드 작성 및 실행해보기

1) test 코드 파일들 수정 작업!

  • 목적:
    • test_main.py에는 "/" path 검증하는 코드만 두기
    • test_todos_api.py에는 기존 todos 관련 검증하는 코드 두기
  • 기존 test_main.py의 이름을 test_todos_api.py로 수정

  • 새 test_main.py 생성

  • test_todos_api.py에 있던 def test_health_check(client)를 test_main.py에 옮기기

 

  • 이제 test_main.py에는 "/" path만 검증하는 테스트 코드가 작성되어있고, test_todos_api.py에는 ToDos 관련된 path들만 검증하는 테스트 코드가 작성되어있음

2) test_users_api.py 생성 및 작성하기

  • src/tests 디렉토리 안에 test_users_api.py 파일 생성하기

# /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
  • test_user_sign_up() 이라는 이름으로 함수 정의, fixture로 client 주기 (client는 conftest.py에 정의한 fixture)
  • response 객체 생성: client에 post 메소드를 사용해서 "/users/sign-up"으로 요청을 보내보는 것
  • assert를 통해 status_code가 200이면 통과, 아니면 강제 에러 발생시킴
  • assert를 통해 response가 json 형태가 되면 통과, 아니면 강제 에러 발생시킴

3) test_users_api.py 실행해보기

  • 터미널에서 $ pytest tests/test_users_api.py 실행

  • 일단 API 연결은 잘 됐다는 것 확인

5. user.py 수정

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

router = APIRouter(prefix="/users")

@router.post("/sign-up", status_code=201)	# 상태코드 추가
def user_sign_up_handler():
    # 
    return True

6. bcrypt 로 암호화 실습 (Python Console)

  • 우리가 비밀번호를 받을 때 비밀번호를 hash해서 개발자들도 알아보지 못하게 할 것

1) bcrypt 라이브러리 설치

  • 라이브러리 설치: bcrypt (암호화 라이브러리)

 

2) bcrypt의 salt 사용하여 비밀번호 암호화 

  • 보통 해시하면 똑같은 해시 값이 나오는데 bcrypt의 salt를 사용하면 매번 다른 hash값이 나옴
>>> import bcrypt
>>> password = "password"
>>> byte_password = password.encode("UTF-8")
>>> hash_1 = bcrypt.hashpw(byte_password, salt=bcrypt.gensalt())
>>> hash_2 = bcrypt.hashpw(byte_password, salt=bcrypt.gensalt())
>>> byte_password
b'password'
>>> hash_1
b'$2b$12$8YD.OgeTIg3FNJEUNb8kZ.CYczovMj1KV1EhiXngdgl2mcYeJL9cG'
>>> hash_2
b'$2b$12$5g4ybQC9uasrp2iwqAdfs.Yu057CtfNSuD517x7pqNOYzkxYkaLs.'

  • import bcrypt
    • bcrypt는 비밀번호를 해시하고 검증하기 위해 사용되는 암호화 라이브러리
    • 이 라이브러리를 사용해 비밀번호 해싱을 수행하고 해시된 비밀번호는 안전하게 저장할 수 있음
  • password = "password"
    • password라는 변수에 평문 비밀번호 "password"가 저장됨
    • 일반 텍스트로, 나중에 안전하게 저장하기 위해 암호화(hashing)할 것
  • byte_password = password.encode("UTF-8")
    • 평문 비밀번호 "password"는 문자열 형태로 저장되어 있는데, bcrypt는 문자열이 아닌 바이트(byte) 데이터를 처리함
    • password.encode("UTF-8")을 사용해 password 문자열을 UTF-8 인코딩 방식으로 바이트 문자열로 변환
    • 이렇게 하면 password는 b'password'로 변환됨
  • hash_1 = bcrypt.hashpw(byte_password, salt=bcryte.gensalt())
    • 비밀번호 해싱이 수행됨
    • bcrypt.hashpw()는 주어진 바이트 형태의 비밀번호(byte_password)와 salt를 사용해 해시를 생성함
    • bcrypt.gensalt()는 솔트 값을 생성하는 함수. 솔트는 비밀번호 해싱 시 해커가 동일한 비밀번호에 대해 동일한 해시 값을 얻게 되느 ㄴ것을 방지하기 위해 사용됨
      • 솔트는 랜덤 값이기에 같은 비밀번호라도 매번 다른 해시값이 생성됨
    • 결과적으로, hash_1 변수에는 해싱된 비밀번호가 저장
  • 실행 과정:
    • bcrypt의 hashpw() 메소드에 byte_password를 전달하고 salt 키워드에 bcrypt.gensalt()를 넣어줘서 hash 생성 
    • 이 해시 값은 비밀번호 복원이 불가능하며, bcrypt를 사용해 저장된 해시와 입력 비밀번호의 해시를 비교하는 방식으로 비밀번호 검증이 이루어짐

3) bcrypt의 checkpw 사용하여 생성한 해시 검증

  • bcrypt.checkpw()를 사용하여 평문 비밀번호를 해시된 비밀번호와 비교하여 비밀번호가 일치하는지 확인 
>>> bcrypt.checkpw(byte_password, hash_1)
True
>>> bcrypt.checkpw(byte_password, hash_2) 
True
  • bcrypt.checkpw(byte_password, hash_1)
    • 바이트 문자열 형태인 byte_password (= b'password')와 생성한 해시된 비밀번호인 hash_1를 입력받아 확인하는 것
  • True 
    • byte_password로 해시된 값이 hash_1과 일치한다는 것을 의미, 즉 평문 비밀번호가 정확하다는 뜻
  • bcrypt.checkpw(byte_password, hash_2)
    • 바이트 문자열 형태인 byte_password (= b'password')와 생성한 해시된 비밀번호인 hash_2를 입력받아 확인하는 것
  • True
    • 비록 hash_2가 hash_1과 다른 값일지라도, 두 해시 모두 동일한 byte_password에서 파생되었기 때문에 결과적으로 True가 반환된 것
  • 실행과정:
    • bcrypt.checkpw(byte_password, hash_1)는 내부적으로 byte_password를 다시 해싱한 후, 그 결과를 hash_1과 비교합니다. 두 값이 동일하면 True를 반환하고, 다르면 False를 반환.
    • bcrypt.checkpw(byte_password, hash_2)는 해시값이 다를 수 있지만, 솔트가 다르더라도 동일한 평문 비밀번호로부터 생성된 해시라면 여전히 일치, 그러므로 True 반환.
    • bcrypt는 솔트를 내부적으로 관리하면서, 저장된 해시값에서 솔트를 추출해 평문 비밀번호를 해시한 결과를 비교. 이 과정에서 두 값이 일치하면 True를 반환
    • 즉, 같은 비밀번호라도 솔트가 다르면 해시값이 달라질 수 있지만, bcrypt는 솔트가 포함된 해시값을 저장하고 있기 때문에 이를 올바르게 비교 가능 

Moving Foward

  • 다음 실습에서는 bcrypt.hashpw()와 bcrypt.checkpw를 사용하여 비밀번호 암호화 및 확인 로직을 구현할 것

 

이때까지의 코드들: 12개

  • /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/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

│     ├── 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)
# /c/Users/관리자/Desktop/projects/todos/src/api/user.py 내용
from fastapi import APIRouter

router = APIRouter(prefix="/users")

@router.post("/sign-up", status_code=201)
def user_sign_up_handler():
    
    return True
# /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/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")
# /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
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()
# /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]
# /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)
  • test_main.py 수정
# /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"}
  • test_todos_api.py 생성
# /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"}
  • test_users_api.py 생성
# /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 실습6 회원가입 API 테스트  (2) 2024.11.16
5 실습5 회원가입 API 구현  (0) 2024.10.18
5 기능 고도화 Lazy Loading / Eager Loading  (0) 2024.10.18
5 실습3 ORM JOIN  (1) 2024.10.16
5 실습2 기능 고도화 User 테이블 생성  (0) 2024.08.15

관련글

티스토리툴바