앞 단계 참조 링크:
- 상태 코드: 2024.03.26 - [Web 개발/FAST API (인프런 강의 내용)] - 1 FastAPI 알아보기
- 프로젝트 소개 및 환경 구축: 2024.04.05 - [Web 개발/FAST API (인프런 강의 내용)] - 1 실습1 GET API 전체조회
- 1 실습1 GET API ToDo 전체 조회: 2024.04.05 - [Web 개발/FAST API (인프런 강의 내용)] - 1 실습1 GET API 전체조회
- 1 실습2 GET API ToDo 단일 조회: 2024.04.09 - [Web 개발/FAST API (인프런 강의 내용)] - 1 실습2 GET API 단일조회
- 1 실습3 POST API ToDo 생성: 2024.04.15 - [Web 개발/FAST API (인프런 강의 내용)] - 1 실습3 POST API todo 생성
- 1 실습4 PATCH API ToDo 수정: 2024.04.16 - [Web 개발/FAST API (인프런 강의 내용)] - 1 실습4 PATCH API todo 수정
- 1 실습5 DELETE API ToDo 삭제: 2024.04.17 - [Web 개발/FAST API (인프런 강의 내용)] - 1 실습5 DELETE API todo 삭제
- 1 실습6 ERROR 처리: 2024.04.18 - [Web 개발/FAST API (인프런 강의 내용)] - 1 실습6 ERROR 처리
- 2 데이터베이스: 2024.04.24 - [Web 개발/FAST API (인프런 강의 내용)] - 2 데이터베이스
- 2 실습1 MySQL 컨테이너 실행 (docker): 2024.04.24 - [Web 개발/FAST API (인프런 강의 내용)] - 2 실습1 MySQL 컨테이너 실행 (docker)
- 2 실습2 MySQL 접속 및 사용: 2024.04.25 - [Web 개발/FAST API (인프런 강의 내용)] - 2 실습2 MySQL 접속 및 사용
- 2 실습3 데이터베이스 연결: 2024.04.25 - [Web 개발/FAST API (인프런 강의 내용)] - 2 실습3 데이터베이스 연결
- 2 실습4 ORM 모델링: 2024.05.02 - [Web 개발/FAST API (인프런 강의 내용)] - 2 실습4 ORM 모델링
- 2 실습5 ORM GET 전체조회 API: 2024.05.03 - [Web 개발/FAST API (인프런 강의 내용)] - 2 실습5 ORM 적용 - GET 전체조회 API
- 2 실습6 ORM HTTP Response 처리: 2024.05.10 - [Web 개발/FAST API (인프런 강의 내용)] - 2 실습6 ORM 적용 - HTTP Response 처리
- 2 실습7 ORM GET 단일조회 API: 2024.05.14 - [Web 개발/FAST API (인프런 강의 내용)] - 2 실습7 ORM 적용 - GET 단일조회 API
- 2 실습8 ORM Refactoring: 2024.06.01 - [Web 개발/FAST API (인프런 강의 내용)] - 2 실습8 ORM 적용 - Refactoring
- 2 실습9 ORM POST API: 2024.06.01 - [Web 개발/FAST API (인프런 강의 내용)] - 2 실습9 ORM 적용 - POST API
- 2 실습10 ORM PATCH API: 2024.06.01 - [Web 개발/FAST API (인프런 강의 내용)] - 2 실습10 ORM 적용 - PATCH API
- 2 실습11 ORM DELETE API: 2024.06.01 - [Web 개발/FAST API (인프런 강의 내용)] - 2 실습11 ORM 적용 - DELETE API
- 3 테스트 코드 PyTest: 2024.06.01 - [Web 개발/FAST API (인프런 강의 내용)] - 3 테스트 코드 PyTest
- 3 실습1 PyTest 세팅: 2024.07.22 - [Web 개발/FAST API (인프런 강의 내용)] - 3 실습1 PyTest 세팅
- 3 실습2 테스트 코드 GET 전체조회 API: 2024.08.01 - [Web 개발/FAST API (인프런 강의 내용)] - 3 실습2 테스트 코드 - GET 전체조회 API
- 3 실습3 테스트 코드 PyTest Mocking: 2024.08.02 - [Web 개발/FAST API (인프런 강의 내용)] - 3 실습3 테스트 코드 - PyTest Mocking
- 3 실습4 테스트 코드 PyTest Fixture: 2024.08.03 - [Web 개발/FAST API (인프런 강의 내용)] - 3 실습4 테스트 코드 - PyTest Fixture
- 3 실습5 테스트 코드 GET 단일조회 API: 2024.08.05 - [Web 개발/FAST API (인프런 강의 내용)] - 3 실습5 테스트 코드 - GET 단조회 APIsf
- 3 실습6 테스트 코드 POST API: 2024.08.05 - [Web 개발/FAST API (인프런 강의 내용)] - 3 실습6 테스트 코드 - POST API
- 3 실습7 테스트 코드 PATCH API: 2024.08.05 - [Web 개발/FAST API (인프런 강의 내용)] - 3 실습7 테스트 코드 - PATCH API
- 3 실습8 테스트 코드 DELETE API: 2024.08.07 - [Web 개발/FAST API (인프런 강의 내용)] - 3 실습8 테스트 코드 - DELETE API
- 4 실습1 Refactoring FastAPI Router: 2024.08.08 - [Web 개발/FAST API (인프런 강의 내용)] - 4 실습1 Refactoring - FastAPI Router
- 4 실습2 Refactoring 의존성 주입: 2024.08.08 - [Web 개발/FAST API (인프런 강의 내용)] - 4 실습2 Refactoring - Dependency Injection 의존성 주입
- 4 실습3 Refactoring 레포지토리 패턴: 2024.08.08 - [Web 개발/FAST API (인프런 강의 내용)] - 4 실습3 Refactoring - Repository Pattern 레포지토리 패
- 5 기능 고도화 SQL JOIN: 2024.08.11 - [Web 개발/FAST API (인프런 강의 내용)] - 5 기능 고도화 SQL JOINㄹ
- 5 실습1 User 테이블 모델링: 2024.08.12 - [Web 개발/FAST API (인프런 강의 내용)] - 5 실습1 기능 고도화 User 모델링
- 5 실습2 User 테이블 생성: 2024.08.15 - [Web 개발/FAST API (인프런 강의 내용)] - 5 실습2 기능 고도화 User 테이블 생성
- 5 실습3 ORM JOIN: 2024.10.16 - [Web 개발/FAST API (인프런 강의 내용)] - 5 실습3 ORM JOIN
- 5 기능 고도화 Lazy Loading / Eager Loading: 2024.10.18 - [Web 개발/FAST API (인프런 강의 내용)] - 5 기능 고도화 Lazy Loading / Eager Loading
- 5 실습4 회원가입 API 생성 & 비밀번호 암호화(bcrypt): 2024.10.18 - [Web 개발/FAST API (인프런 강의 내용)] - 5 실습4 회원가입 API 생성 & 비밀번호 암호화(bcrypt)
- 5 실습5 회원가입 API 구현: 2024.10.18 - [Web 개발/FAST API (인프런 강의 내용)] - 5 실습5 회원가입 API 구현
- 5 실습6 회원가입 API 테스트: 2024.11.16 - [분류 전체보기] - 5 실습6 회원가입 API 테스트
- 5 기능 고도화 로그인 / 유저 인증 JWT: 2024.11.16 - [Web 개발/FAST API (인프런 강의 내용)] - 5 기능 고도화 - 로그인 / 유저 인증 - JWT
- 5 실습7 로그인 API 구현: 2024.11.16 - [Web 개발/FAST API (인프런 강의 내용)] - 5 실습7 로그인 API 구현
00 개요
- 테스트 코드가 반복적인 부분이 있기 때문에 이번에는 테스트 코드 작성 대신 Swagger UI를 통해 호출해보는 실습 진행
- 우선적으로 DB에 있는 비밀번호를 업데이트 해야됨
01 DB 접속 및 사용자 정보 확인
1. docker의 컨테이너 목록 확인
# 실행, 미실행 중인 컨테이너 모두 출력
# docker ps -a
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
0c3283e97292 mysql:8.0 "docker-entrypoint.s…" 7 months ago Exited (255) 3 minutes ago 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
# 미실행 중인 컨테이너만 출력
# docker ps -a --filter 'status=exited'
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
0c3283e97292 mysql:8.0 "docker-entrypoint.s…" 7 months ago Exited (255) 3 minutes ago 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
2. todos 컨테이너 실행
# docker start 컨테이너명
# docker start todos
todos
# 실행 확인
# docker ps
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
0c3283e97292 mysql:8.0 "docker-entrypoint.s…" 7 months ago Up 5 minutes 0.0.0.0:3306->3306/tcp, 33060/tcp todos
3. todos 컨테이너 접속 및 bash 쉘에서 실행
# docker exec -it todos bash
bash-4.4#
4. DB (MySQL) 접속
bash-4.4# mysql -u 사용자 -p
bash-4.4# mysql -u root -p
- root 사용자를 통해 mysql에 접속하게 하는 명령어
- -p 옵션 뒤에 입력 안하고 enter 누를 시 비밀번호를 따로 입력해주겠다는 뜻
- enter 누를 시 'Enter password:'라고 뜸 - 비밀번호(todos) 입력 후 enter 누르면 mysql에 접속하게 됨
bash-4.4# mysql -u root -p
Enter password: # 'todos' 입력
Welcome to the MySQL monitor. Commands end with ; or \g.
Your MySQL connection id is 10
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>
5. 데이터베이스 조회 및 todos 데이터베이스로 이동 (사용)
mysql> show databases;
+--------------------+
| Database |
+--------------------+
| information_schema |
| mysql |
| performance_schema |
| sys |
| todos |
+--------------------+
5 rows in set (0.01 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
6. 테이블 확인 및 사용자 테이블 (user) 조회
mysql> show tables;
+-----------------+
| Tables_in_todos |
+-----------------+
| todo |
| user |
+-----------------+
2 rows in set (0.00 sec)
mysql> select * from user;
+----+--------------+--------------------------------------------------------------+
| id | username | password |
+----+--------------+--------------------------------------------------------------+
| 1 | admin | password |
| 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)
- 강사님의 DB에는 admin 계정만 있는데 나는 테스트 코드로 여러번 해봐서 그런지, 왜인지 잘 모르겠지만 여러 사용자 데이터가 있네..?
- 강의에서는 admin의 비밀번호를 hashed된 password로 수정하는 실습 진행함
7. Python Console에서 hash 생성
# python
Python 3.13.0 (tags/v3.13.0:60403a5, Oct 7 2024, 09:38:07) [MSC v.1941 64 bit (AMD64)] on win32
Type "help", "copyright", "credits" or "license" for more information.
>>> import bcrypt
>>> password = 'password'.encode('UTF-8')
>>> hashed_password = bcrypt.hashpw(password, salt=bcrypt.gensalt())
>>> hashed_password.decode('UTF-8')
'$2b$12$CQ4sJqQAPSRI3P.t69AN1eNXHz8huaPLKUYZnc0VOTnPOybMiPVz.'
8. 사용자 테이블 (user)의 admin 계정의 password 해시화된 password로 변경
# 다시 docker의 todos 컨테이너에 접속, bash 쉘 사용
$ docker exec -it todos bash
bash-4.4# mysql -u root -p
Enter password:
Welcome to the MySQL monitor. Commands end with ; or \g.
Your MySQL connection id is 11
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> show databases;
+--------------------+
| Database |
+--------------------+
| information_schema |
| mysql |
| performance_schema |
| sys |
| todos |
+--------------------+
5 rows in set (0.00 sec)
# todos 데이터베이스로 이동
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)
# user 테이블 데이터 확인
mysql> select * from user;
+----+--------------+--------------------------------------------------------------+
| id | username | password |
+----+--------------+--------------------------------------------------------------+
| 1 | admin | password |
| 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)
# user 테이블의 id = 1인 (admin) 계정의 비밀번호 수정
mysql> update user set password = '$2b$12$CQ4sJqQAPSRI3P.t69AN1eNXHz8huaPLKUYZnc0VOTnPOybMiPVz.' where id = 1;
Query OK, 1 row affected (0.02 sec)
Rows matched: 1 Changed: 1 Warnings: 0
# 수정된 테이블 데이터 확인
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)
02 Swagger UI로 확인
1. 터미널에서 서버 실행
# 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 [19276] using WatchFiles
INFO: Started server process [5772]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: 127.0.0.1:56435 - "GET / HTTP/1.1" 200 OK
2. Swagger UI 접속
- 웹 브라우저에 127.0.0.1/docs 입력
3. Login API 호출
- POST /users/log-in > Try it out > Request body에 {'username': 'admin', 'password': 'password'} 입력 > Execute 클릭
- Response body로 상태 코드 200 및 access token이 반환된 것을 확인
이때까지의 코드들: 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)
# /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()
# /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 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
# /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 안남
# /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 실습9 JWT 사용 인증 구현 및 DB 데이터 조회 (0) | 2024.11.26 |
---|---|
5 실습7 로그인 API 구현 (1) | 2024.11.16 |
5 기능 고도화 - 로그인 / 유저 인증 - JWT (0) | 2024.11.16 |
5 실습6 회원가입 API 테스트 (2) | 2024.11.16 |
5 실습5 회원가입 API 구현 (0) | 2024.10.18 |