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

5 실습9 JWT 사용 인증 구현 및 DB 데이터 조회

by yororing 2024. 11. 26.

앞 단계 참조 링크:

00 개요

  • 지금까지 로그인을 통해 access token을 발급 받는 작업까지 진행함
  • 이제는 access token을 API에서 실제 검증하는 작업을 실습할 것
  • access token을 header로 받아 처리하기 위해 todos 전체 조회 API에 access token을 받을 수 있도록 코드를 추가할 것

01 access token 검증 로직 구현

1. src 디렉토리에 security.py 파일 생성 및 작성

  • API에서 dependency로 사용할 수 있는 함수 작성
# /c/Users/관리자/Desktop/projects/todos/src/security.py 내용
from fastapi import Depends, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

def get_access_token(
    auth_header: HTTPAuthorizationCredentials | None = Depends(HTTPBearer(auto_error=False)),
) -> str:
    if auth_header is None:
        raise HTTPException(
            status_code=401,
            detail='Not Authorized',
        )
    return auth_header.credentials
  • 설명:
    • header에 'bearer'라는 형태로 access token이 있나 검증하는 것
    • 없으면 401 에러 강제 발생시킴
    • auth_header(header)가 정상적으로 요청이 오면 auth_header 안에 있는 credentials에 access token이 반환되는 원리
  • 이렇게 생성된 access token을 api/todos.py의 get_todos_handler() API에서 사용할 것 

2. api/todos.py 수정

  • get_todos_handler()의 인자값으로 token 값을 Depends()로 추가
# /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
from src.security import get_access_token                   # 추가

router = APIRouter(prefix='/todos')

# GET Method 사용하여 전체 조회 API
@router.get("", status_code=200)
def get_todos_handler(
    access_token: str = Depends(get_access_token),	      # 추가
    order: str | None = None,
    todo_repo: ToDoRepository = Depends(),
) -> ToDoListSchema:

    print("==========")                   # 추가 (확인용, 추후 제거할 것)
    print(access_token)                   # 추가
    print("==========")                   # 추가
    
    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_todos_handler() API가 호출될 때마다 get_access_token이라는 dependency가 먼저 호출이 되어 header에 'bearer'의 형태로 access token이 있나 먼저 검증을 수행한 후 access token이 없을 경우 401 에러 강제 발생시킴
  • header (auth_header)가 정상적으로 요청이 오면 그 안에 있는 credentials에서 access token이 반환됨
  • 이렇게 생성된 access token을 get_todos_handler() API에서 사용할 수 있게 됨

3. database/connection.py 수정

  • access token을 사용하기 위해 database/connection.py에서 engine = create_engine(DATABASE_URL, echo=True)에서 'echo=True' 옵션을 제거할 것 → engine = create_engine(DATABASE_URL)
    • 이유: DB에서 쿼리가 계속 발생하면 로그를 보기 힘듦 
# /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,         # 제거: DB에서 쿼리가 계속 발생하면 로그를 보기 힘들기 때문
                       )
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)

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

02 Swagger UI에서 확인

1. HTTP Bearer 인증 사용하기

  • 터미널에서 서버 실행
# cd src
# uvicorn main:app --reload
INFO:     Will watch for changes in these directories: ['C:\\Users\\관리자\\Desktop\\projects\\todos\\src']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [14736] using WatchFiles
INFO:     Started server process [18380]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
  • 웹 브라우저에 127.0.0.1:8000/docs 입력하여 Swagger UI로 이동

  • 전과 달리 오른쪽 윗부분에 'Authorize ' 버튼이 생성된 것을 확인
  • Authorize 클릭 > HTTPBearer를 헤더에 Value로 입력 가능 

  • 일단 access token을 발급 받기 위해 로그인 하기

2. 로그인을 통해 Access Token 받기

  • POST /users/log-in > Try it out > Request body: { "username": "admin", "password": "password" } 입력 > Execute

  • 500 에러: Internal Server Error
    • 이유: 데이터베이스를 조회해야 하는데 MySQL이 담겨 있는 docker가 실행중이지 않아서임

  • 해결 방법: docker 실행
  • windows에서 docker (나는 Docker Desktop) 앱을 클릭하여 띄우기 전에는 터미널에서 docker 명령어 안됨

  • 앱을 클릭하여 띄운 후에는 터미널에서 docker 명령어 사용가능함

  • docker 명령어로 todos 컨테이너 실행하기
# todos 컨테이너 시작
# docker start todos
todos

# 모든 컨테이너 상태 확인
# docker ps -a
CONTAINER ID   IMAGE                             COMMAND                   CREATED        STATUS                    PORTS                               NAMES
0c3283e97292   mysql:8.0                         "docker-entrypoint.s…"    7 months ago   Up 2 seconds              0.0.0.0:3306->3306/tcp, 33060/tcp   todos
2597a124c27d   docker/welcome-to-docker:latest   "/docker-entrypoint.…"    7 months ago   Exited (0) 7 months ago                                       welcome-to-docker

# 실행 중인 컨테이너 상태 확인
# docker ps
CONTAINER ID   IMAGE       COMMAND                   CREATED        STATUS         PORTS                               NAMES
0c3283e97292   mysql:8.0   "docker-entrypoint.s…"    7 months ago   Up 2 seconds   0.0.0.0:3306->3306/tcp, 33060/tcp   todos
  • 다시 Swagger UI에서 로그인 시도해보기: Reset > Request body: { "username": "admin", "password": "password" } 입력 > Execute

  • Server response에 200 응답과 함께 access token이 발급된 것을 확인

  • access_token 값을 복사하여 Authorize > HTTPBearer (http, Bearer)의 Value: 에 붙여넣기 > Authorize 클릭

  • 이렇게 하면 모든 요청에 해당 header를 사용하게 됨
  • 전체 todo API를 조회 해보면 Try it out > Execute 클릭 시 헤더에 Authorization이 Bearer 형태로 JWT access token이 같이 전달 된 것을 확인
    • -H 'Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhZG1pbiIsImV4cCI6MTczMjcwMjM2Nn0.S1NOTgZaGUbWOWc2r7jwBfgj3vtrTruOoe3FF5FAPiw'
    • curl -X 'GET' \
  'http://127.0.0.1:8000/todos' \
  -H 'accept: application/json' \
  -H 'Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhZG1pbiIsImV4cCI6MTczMjcwMjM2Nn0.S1NOTgZaGUbWOWc2r7jwBfgj3vtrTruOoe3FF5FAPiw'

  • 문제 없이 응답에도 성공한 것을 확인
    • Server response: 200

  • 터미널로 가보면 access_token이 출력된 것도 확인

3. 로그아웃 후 API 호출해보기

  • Swagger UI로 가서 Authorize > Logout > Close > GET /todos > (Cancel) > Try it out > Execute

 

  • 우리가 dependency로 걸었던 401 에러 발생하면서 데이터 조회 불가능한 것을 확인 

03 access token 검증 로직 구현 (continue)

0. JWT decode 해보기

  • 먼저 Python Console에서 JWT를 decoding하는 실습 진행해보기
# python
Python 3.12.1 (tags/v3.12.1:2305ca5, Dec  7 2023, 22:03:25) [MSC v.1937 64 bit (AMD64)] on win32
Type "help", "copyright", "credits" or "license" for more information.
>>> from jose import jwt
>>> access_token = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhZG1pbiIsImV4cCI6MTczMjcwMjM2Nn0.S1NOTgZaGUbWOWc2r7jwBfgj3vtrTruOoe3FF5FAPiw' 
>>> secret_key: str = "a0ef6b27d5050d0040867160134b69bd673f640dd1db8db8b0ee0166114f4cfa"
>>> jwt_algorithm: str = "HS256"
>>> jwt.decode(access_token, secret_key, algorithms=[jwt_algorithm])
{'sub': 'admin', 'exp': 1732702366}

  • access token을 decode하면 우리가 'sub'(subject)로 넣었던 admin (username)과 'exp'(만료 기간)이 반환되는 것을 확인
  • 이렇게 access token을 decoding하여 반환되는 값 ('sub')을 사용할 것

1. service/user.py에 추가

  • decode_jwt()라는 함수 작성: access token을 받아 decoding하여 username을 반환하는 기능 구현
# /c/Users/관리자/Desktop/projects/todos/src/service/user.py 내용
from datetime import datetime, timedelta
from jose import jwt
import bcrypt
import logging

class UserService:
    encoding: str = "UTF-8"
    secret_key: str = "a0ef6b27d5050d0040867160134b69bd673f640dd1db8db8b0ee0166114f4cfa"
    jwt_algorithm: str = "HS256"

...

    def decode_jwt(self, access_token: str) -> str:      # 추가
        payload: dict = jwt.decode(
            access_token, self.secret_key, algorithms=[self.jwt_algorithm]
        )
        # expire: 원래는 여기서 토큰의 만료 기간을 검증하는 부분 필요하나 우리는 거기까지 안 할 것임
        return payload['sub'] # username

2. api/todos.py 수정

  • get_todos_handler()에서 access token을 검증하고 username을 꺼내보는 작업 진행할 것
  • 변경되는 내용:
    • 1. access token을 통해 사용자가 인증됐다는 것을 확인 후
    • 2. access token에 담긴 username을 사용하여 DB에서 사용자의 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, UserRepository, User # 추가
from src.schema.response import ToDoSchema, ToDoListSchema
from src.schema.request import CreateToDoRequest
from src.security import get_access_token
from src.service.user import UserService                            # 추가

router = APIRouter(prefix='/todos')

# GET Method 사용하여 전체 조회 API
@router.get("", status_code=200)
def get_todos_handler(
    access_token: str = Depends(get_access_token),	    # 추가
    order: str | None = None,
    user_service: UserService = Depends(),              # 추가
    user_repo: UserRepository = Depends(),
    # todo_repo: ToDoRepository = Depends(),             # 삭제
) -> ToDoListSchema:
    username: str = user_service.decode_jwt(access_token=access_token)  # 추가, 이 username을 통해 사용자를 조회할 것
    user: User | None = user_repo.get_user_by_username(username=username)  # 추가, 이 user_id를 통해 User(테이블)를 불러와 User에 담긴 ToDo를 조회할 것
    if not user: # 회원 탈퇴, etc. 경우
        raise HTTPException(status_code=404, detail='User Not Found')
    # todos: List[ToDo] = todo_repo.get_todos()         # 삭제
    todos: List[ToDo] = user.todos                      # User에 있는 ToDo들을 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]

 

04 Swagger UI에서 확인

1. HTTP Bearer에 담긴 access token에 담긴 username 사용하여 ToDo 목록 조회해보기

  • 먼저 admin의 todo 목록 DB에서 확인
# todos 컨테이너를 열어 bash에서 실행

# docker exec -it todos bash
bash-4.4#
# mysql 접속

bash-4.4# mysql -u root -p 
Enter password: 
Welcome to the MySQL monitor.  Commands end with ; or \g.
Your MySQL connection id is 12
Server version: 8.0.36 MySQL Community Server - GPL

Copyright (c) 2000, 2024, Oracle and/or its affiliates.

Oracle is a registered trademark of Oracle Corporation and/or its
affiliates. Other names may be trademarks of their respective
owners.

Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.

mysql>
# DB 확인 및 todos DB로 이동
mysql> show databases;
+--------------------+
| Database           |
+--------------------+
| information_schema |
| mysql              |
| performance_schema |
| sys                |
| todos              |
+--------------------+
5 rows in set (0.03 sec)

mysql> use todos
Reading table information for completion of table and column names
You can turn off this feature to get a quicker startup with -A

Database changed
# 테이블 확인
mysql> show tables
    -> ;
+-----------------+
| Tables_in_todos |
+-----------------+
| todo            |
| user            |
+-----------------+
2 rows in set (0.00 sec)

# admin의 id 확인 (=1)
mysql> select * from user;
+----+--------------+--------------------------------------------------------------+
| id | username     | password                                                     |
+----+--------------+--------------------------------------------------------------+
|  1 | admin        | $2b$12$CQ4sJqQAPSRI3P.t69AN1eNXHz8huaPLKUYZnc0VOTnPOybMiPVz. |
|  2 | test         | $2b$12$cdBayNyDQ3dHy4tuxTRS6.UjS31fTZazhikxxb1b4vpl9XSE/L0Ke |
|  6 | test1        | $2b$12$gX5hZR0uC1VWvJ7wbhn6SOxHT586q260Y9R6FsByfQ.2yQTYCuItW |
|  7 | test2        | $2b$12$G67XckbQKeDkC.TvJiGU9uPrR0wSJ0yDXhdvBUpGkcDE6vXFeXRgm |
|  9 | test5        | $2b$12$ocN5JefdnIPDOKHbS4Vr8uS3D3M/b5dGnv0SqGx75soCvWoooGX52 |
| 11 | test3        | $2b$12$tlFyh7rYIQAi/mW1OemevO43gVwOC/uWGnISqR9z9xuczf9Gg/WzG |
| 13 | reawsdfsdfad | $2b$12$xEKfydJ.sXmkDMvzqviBCO9.yvXKouYqV1cx/vTNEy9NguGebP71y |
| 15 | sdfg         | $2b$12$6fiHfI.OK67pEwM3eKx.LupSmc/WaftsMTO7JaWFdaImdtt1rPGYq |
| 16 | test7        | $2b$12$LC2cov74JEIXoWpc0Ee.bOLvM170efkAsdZX27o5p7wNqqbriAxCK |
| 17 | test71       | $2b$12$Ms.IAep6AoGxdxL7MZUIneWV.K6RF7wItk8gO.q1oXdJ12.ph4iku |
| 18 | test11       | $2b$12$UgoZpQQwvXvfS7UdpJs68Oz9qeE3O.0hEPXnwAfeTXjulr3k0/wYy |
| 21 | ertet        | $2b$12$8e/nQcJNTWTAF425vu0Ns.4Io9z849jKGjM8Vd9ME3gaNmqwaviPC |
+----+--------------+--------------------------------------------------------------+
12 rows in set (0.00 sec)

# id = user_id = 1의 todo 목록 확인 (todo의 user_id는 user의 id랑 연결되어있음)
mysql> select * from todo;
+----+-------------------+---------+---------+
| id | contents          | is_done | user_id |
+----+-------------------+---------+---------+
|  1 | FastAPI Section 0 |       1 |       1 |
|  2 | FastAPI Section 2 |       1 |       1 |
|  3 | FastAPI Section 3 |       1 |       1 |
+----+-------------------+---------+---------+
3 rows in set (0.00 sec)
  • →  admin의 todos는 위의 3개 전체 다 인것을 확인
  • Swagger UI에서 Authorize > HTTPBearer (http, Bearer)의 Value: 에 복사한 access_token 붙여넣기 ('eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhZG1pbiIsImV4cCI6MTczMjcwMjM2Nn0.S1NOTgZaGUbWOWc2r7jwBfgj3vtrTruOoe3FF5FAPiw') > Authorize 클릭

 

  • GET /todos > Try it out > Execute 클릭

  • admin의 todos 목록 (3개 다)들 반환되는 것을 확인 
  • DB에서 admin의 todo 목록 변경해보기
    • todo 테이블의 id=3인 todo 데이터의  user_id를 2로 변경 (admin의 user_id는 1)
# todo 테이블의 id=3인 todo 데이터의  user_id를 2로 변경 및 저장 (commit)

mysql> UPDATE todo SET user_id = 2 WHERE id = 3;
Query OK, 1 row affected (0.02 sec)
Rows matched: 1  Changed: 1  Warnings: 0

mysql> select * from todo;
+----+-------------------+---------+---------+
| id | contents          | is_done | user_id |
+----+-------------------+---------+---------+
|  1 | FastAPI Section 0 |       1 |       1 |
|  2 | FastAPI Section 2 |       1 |       1 |
|  3 | FastAPI Section 3 |       1 |       2 |
+----+-------------------+---------+---------+
3 rows in set (0.00 sec)

mysql> commit;
Query OK, 0 rows affected (0.00 sec)
  • Swagger UI에서 변경사항 확인해보기:
    • Authorize > HTTPBearer (http, Bearer)의 Value: 에 복사한 access_token 붙여넣기 ('eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhZG1pbiIsImV4cCI6MTczMjcwMjM2Nn0.S1NOTgZaGUbWOWc2r7jwBfgj3vtrTruOoe3FF5FAPiw') > Authorize 클릭
    • GET /todos > Try it out > Execute 클릭

  • admin의 todo 목록 (2개)만 조회되는 것을 확인! 성공! 

2. api/todos.py 수정

  • 그럼 이제 더 이상 api/todos.py의 get_todos_handler()는 DB에 저장된 모든 todo 목록들을 조회하는 것이 아닌 특정 사용자 (username)의 todo들만 조회하는 API가 된 것!
  • 그래서 주석 바꿔줌
# /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, UserRepository, User # 추가
from src.schema.response import ToDoSchema, ToDoListSchema
from src.schema.request import CreateToDoRequest
from src.security import get_access_token
from src.service.user import UserService                # 추가

router = APIRouter(prefix='/todos')

# GET Method 사용하여 username의 todo 조회 API          # 변경 (원래: GET Method 사용하여 전체 조회 API)
@router.get("", status_code=200)
def get_todos_handler(
    access_token: str = Depends(get_access_token),	    # 추가
    order: str | None = None,
    user_service: UserService = Depends(),              # 추가
    user_repo: UserRepository = Depends(),
    # todo_repo: ToDoRepository = Depends(),             # 삭제
) -> ToDoListSchema:
    username: str = user_service.decode_jwt(access_token=access_token)  # 추가, 이 username을 통해 사용자를 조회할 것
    user: User | None = user_repo.get_user_by_username(username=username)  # 추가, 이 user_id를 통해 User(테이블)를 불러와 User에 담긴 ToDo를 조회할 것
    if not user: # 회원 탈퇴, etc. 경우
        raise HTTPException(status_code=404, detail='User Not Found')
    # todos: List[ToDo] = todo_repo.get_todos()         # 삭제
    todos: List[ToDo] = user.todos                      # User에 있는 ToDo들을 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]
    )

05 테스트 코드 수정

1. tests/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, User                             # 추가
from src.database.repository import ToDoRepository, UserRepository  # 추가
from src.service.user import UserService                            # 추가

# 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 사용하여 username의 todo 목록 조회 API 검증                # 추가
def test_get_todos(client, mocker):
    access_token: str = UserService().create_jwt(username="test")
    headers = {'Authorization': f"Bearer {access_token}"}
    user = User(id=1, username="test", password="hashed")
    user.todos = [
        ToDo(id=1, contents="FastAPI Section 0", is_done=True),
        ToDo(id=2, contents="FastAPI Section 1", is_done=False),
    ]
    mocker.patch.object(
        UserRepository, "get_user_by_username", return_value=user
    )
    # order = ASC
    response = client.get('/todos', header=headers)
    assert response.status_code == 200

...

 

이때까지의 코드들: 14개

  • /c/Users/관리자/Desktop/projects/todos/src/main.py
  • /c/Users/관리자/Desktop/projects/todos/src/security.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

│     └──  security.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"}
  • security.py 추가
# /c/Users/관리자/Desktop/projects/todos/src/security.py 내용
from fastapi import Depends, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

def get_access_token(
    auth_header: HTTPAuthorizationCredentials | None = Depends(HTTPBearer(auto_error=False)),
) -> str:
    if auth_header is None:
        raise HTTPException(
            status_code=401,
            detail='Not Authorized',
        )
    return auth_header.credentials
  • api/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, UserRepository, User # 추가
from src.schema.response import ToDoSchema, ToDoListSchema
from src.schema.request import CreateToDoRequest
from src.security import get_access_token
from src.service.user import UserService                # 추가

router = APIRouter(prefix='/todos')

# GET Method 사용하여 username의 todo 조회 API          # 변경 (원래: GET Method 사용하여 전체 조회 API)
@router.get("", status_code=200)
def get_todos_handler(
    access_token: str = Depends(get_access_token),	    # 추가
    order: str | None = None,
    user_service: UserService = Depends(),              # 추가
    user_repo: UserRepository = Depends(),
    # todo_repo: ToDoRepository = Depends(),             # 삭제
) -> ToDoListSchema:
    username: str = user_service.decode_jwt(access_token=access_token)  # 추가, 이 username을 통해 사용자를 조회할 것
    user: User | None = user_repo.get_user_by_username(username=username)  # 추가, 이 user_id를 통해 User(테이블)를 불러와 User에 담긴 ToDo를 조회할 것
    if not user: # 회원 탈퇴, etc. 경우
        raise HTTPException(status_code=404, detail='User Not Found')
    # todos: List[ToDo] = todo_repo.get_todos()         # 삭제
    todos: List[ToDo] = user.todos                      # User에 있는 ToDo들을 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 Body, HTTPException, Depends, APIRouter
from src.schema.request import SignUpRequest, LogInRequest
from src.schema.response import UserSchema, JWTResponse
from src.service.user import UserService
from src.database.orm import User
from src.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로 들어감, 실제 DB에 사용자가 생성됨

    # user(id, username) 값 반환
    return UserSchema.from_orm(user)

@router.post("/log-in")
def user_log_in_handler(
    request: LogInRequest,
    user_repo: UserRepository = Depends(),
    user_service: UserService = Depends(),
    
):
    # 사용자 정보 DB 조회
    user: User | None = user_repo.get_user_by_username(username=request.username)
    if not user:
        raise HTTPException(status_code=404, detail = "User Not Found")
    
    # 비밀번호 검증
    verified: bool = user_service.verify_password(
        plain_password=request.password,
        hashed_password=user.password
    )
    if not verified:
        raise HTTPException(status_code=401, detail = "Not Authorized")
    
    # jwt 생성
    access_token: str = user_service.create_jwt(username=user.username)
    
    # jwt 반환
    return JWTResponse(access_token=access_token)
  • database/connection.py 수정
# /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,         # 제거: DB에서 쿼리가 계속 발생하면 로그를 보기 힘들기 때문
                       )
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")
    
    @classmethod
    def create(cls, username: str, hashed_password: str) -> "User":
        return cls(
            username=username,
            password=hashed_password,
        )

 

# /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 get_user_by_username(self, username: str) -> User | None:
        return self.session.scalar(select(User).where(User.username == username))
    
    def save_user(self, user: User) -> User:
        self.session.add(instance=user)
        self.session.commit()
        self.session.refresh(instance=user)
        return user

 

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

class LogInRequest(BaseModel):
    username: str
    password: str

 

# /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 안남

class JWTResponse(BaseModel):
    access_token: str
  • service/user.py 수정 (decode_jwt() 메소드 추가)
# /c/Users/관리자/Desktop/projects/todos/src/service/user.py 내용
from datetime import datetime, timedelta
from jose import jwt
import bcrypt
import logging

class UserService:
    encoding: str = "UTF-8"
    secret_key: str = "a0ef6b27d5050d0040867160134b69bd673f640dd1db8db8b0ee0166114f4cfa"
    jwt_algorithm: str = "HS256"
    
    def hash_password(self, plain_password: str) -> str:
        '''
        input: plain_password (str)
        output: hashed_password (str)
        function: receives plain-text password and returns hashed password
        '''
        logging.info("hash_password called with: %s", plain_password)
        hashed_password: bytes = bcrypt.hashpw(
            plain_password.encode(self.encoding),
            salt=bcrypt.gensalt()
        )
        return hashed_password.decode(self.encoding)
    
    def verify_password(self, plain_password: str, hashed_password: str) -> bool:
        '''
        input: plain_password (str)
        output: hashed_password (str)
        function: receives plain-text password and hashed password, returns bool whether they are the same password or not
        '''
        return bcrypt.checkpw(
            plain_password.encode(self.encoding),
            hashed_password.encode(self.encoding)
        )
        
    def create_jwt(self, username: dict) -> str:
        return jwt.encode(
            {"sub": username,
             "exp": datetime.now() + timedelta(days=1) },
            self.secret_key,
            algorithm=self.jwt_algorithm
        )
    
    def decode_jwt(self, access_token: str) -> str:       # 추가
        payload: dict = jwt.decode(
            access_token, self.secret_key, algorithms=[self.jwt_algorithm]
        )
        # expire: 원래는 여기서 토큰의 만료 기간을 검증하는 부분 필요하나 우리는 거기까지 안 할 것임
        return payload['sub'] # username

 

# /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"}
  • tests/test_todos_api.py 수정 (test_get_todos() 함수 수정)
# /c/Users/관리자/Desktop/projects/todos/src/tests/test_todos_api.py 내용
from src.schema.response import ToDoSchema
from src.database.orm import ToDo, User                             # 추가
from src.database.repository import ToDoRepository, UserRepository  # 추가
from src.service.user import UserService                            # 추가

# 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 사용하여 username의 todo 목록 조회 API 검증                # 추가
def test_get_todos(client, mocker):
    access_token: str = UserService().create_jwt(username="test")
    headers = {'Authorization': f"Bearer {access_token}"}
    user = User(id=1, username="test", password="hashed")
    user.todos = [
        ToDo(id=1, contents="FastAPI Section 0", is_done=True),
        ToDo(id=2, contents="FastAPI Section 1", is_done=False),
    ]
    mocker.patch.object(
        UserRepository, "get_user_by_username", return_value=user
    )
    # order = ASC
    response = client.get('/todos', header=headers)
    assert response.status_code == 200

# 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 내용
from src.service.user import UserService
from src.database.orm import User
from src.database.repository import UserRepository

username = 'test11'
def test_user_sign_up(client, mocker):                  # mocker는 fixture
    hashed_password = mocker.patch.object(              # UserService.hash_password 메소드에 mocking 적용
        UserService,
        "hash_password",
        return_value="hashed"
    )
    
    user_create = mocker.patch.object(                  # User.create 메소드에 mocking 적용
        User,
        "create",
        return_value=User(id=None, username=username, password="hashed")
    )
    
    mocker.patch.object(                                # UserRepository.save_user 메소드에 mocking 적용
        UserRepository,
        "save_user",
        return_value=User(id=1, username=username, password="hashed")
    )
    
    body = {                                            # API에 넘겨줄 body 추가
        "username": username,
        "password": "plain"
    }
    response = client.post("users/sign-up", json=body)  # json 형식으로 body 넘겨줘서 POST 요청 보내기 추가
    print(hashed_password.call_args_list)
    hashed_password.assert_called_once_with(            # UserService.hash_password 호출 여부 검증
        plain_password="plain"
    )

    user_create.assert_called_once_with(                # User.create 호출 여부 검증
        username=username, hashed_password="hashed"
    )
    
    assert response.status_code == 201          # 201로 변경
    assert response.json() == {"id": 1, "username": username}     # 반환값으로 UserRepository.save_user 검증

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

5 실습10 Redis 컨테이너 실행 & Redis 연결 (docker)  (0) 2025.01.24
5 기능 고도화 Caching  (1) 2024.12.25
5 실습8 로그인 API 테스트  (1) 2024.11.25
5 실습7 로그인 API 구현  (1) 2024.11.16
5 기능 고도화 - 로그인 / 유저 인증 - JWT  (0) 2024.11.16

관련글

티스토리툴바