FAST API에서는 pydantic 의 BaseModel을 이용해서 쉽게 request body 처리 가능
Pydantic 라이브러리란
파이썬의 type annotation을 활용해서 data validataion (데이터 검증)과 setting을 관리해주는 라이브러리
runtime에 type hint를 적용하고 데이터가 invalid할 때 오류 발생시킴
1) 필요한 패키지 참조: pydantic의 BaseModel
# todos > src > main.py
from fastapi import FastAPI
from pydantic import BaseModel # 추가됨
...
2) class 생성
# todos > src > main.py
from fastapi import FastAPI
from pydantic import BaseModel # 추가됨
...
class CreateToDoRequest(BaseModel):
id: int
contents: str
is_done: bool
class CreateToDoRequest(BaseModel):
CreateToDoRequest라는 이름의 클래스 생성
BaseModel을 상속받게 설정
이제, 이 클래스의 request body의 형태를 지정해줘야 함
ToDo item들이 갖는 형태와 동일한 스키마로 이 request를 만들어주도록 지정해주기 위해 다음과 같이 함
id: int
id는 int로 받도록 설정
contents: str
contents는 str으로 받도록 설정
is_done: bool
is_done은 bool으로 받도록 설정
3) POST Method Mapping 및 해당 path에 POST 요청을 보내는 API 생성
# todos > src > main.py
from fastapi import FastAPI
from pydantic import BaseModel # 추가됨
...
@app.post("/todos")
def create_todos_handler(request: CreateToDoRequest):
todo_data[request.id] = request.dict()
return todo_data[request.id]
request인자 생성 후 : CreateToDoRequest 라고 써주면 FastAPI가 알아서 request body를 이 CreateToDoRequest에 넣어 처리해줌 (type 검사 등 여러가지 작업들을 FastAPI가 알아서 해줌)
todo_data[request.id] = request.dict()
기존에 있던 todo_data에 key 값으로 새로운 데이터를 넣어주는 것
key는 request에 있는 id를 그대로 사용할 것
value는 사용자로부터 전달받은 request를 전달해줄 것
주의: request.dict()하는 이유
그냥 request하면 데이터 type이 안 맞아서 에러남
데이터 type이 안 맞는 이유: todo_data는 value가 딕셔너리인데 request는 딕셔너리가 아닌 pydantic의 BaseModel을 상속받은 새로운 클래스의 객체이기에 .dict() 메소드를 사용해서 데이터 type을 딕셔너리로 잡아줘야 됨
→ 사용자로부터 받은 데이터(request)룰 딕셔너리로 바꿔서 todo_data에 넣어주게 됨
return todo_data[request.id]
전달받은 id 값으로 todo_data를 조회하여 해당 todo 내용 반환
2. FastAPI 웹 서버 자동 재시작 (--reload 옵션 사용)
웹 서버 실행 후 코드 변경 시 변경사항을 SwaggerUI에 반영하기 위해선 웹 서버 재시작 (uvicorn 종료 후 다시 실행) 필수
reload 옵션 사용 시 변화가 감지되면 자동으로 서버를 FastAPI가 재시작 됨
방금 수정한 main.py를 저장 (ctrl + S) 하면 자동으로 재시작되어 변경사항이 반영됨
# 수정된 main.py를 저장하면 다음이 터미널에 출력됨
WARNING: StatReload detected changes in 'main.py'. Reloading...
INFO: Shutting down
INFO: Waiting for application shutdown.
INFO: Application shutdown complete.
INFO: Finished server process [4900]
Process SpawnProcess-2:
Traceback (most recent call last):
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\asyncio\runners.py", line 118, in run
return self._loop.run_until_complete(task)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\asyncio\base_events.py", line 684, in run_until_complete
return future.result()
^^^^^^^^^^^^^^^
asyncio.exceptions.CancelledError
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\multiprocessing\process.py", line 314, in _bootstrap
self.run()
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\multiprocessing\process.py", line 108, in run
self._target(*self._args, **self._kwargs)
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\uvicorn\_subprocess.py", line 78, in subprocess_started
target(sockets=sockets)
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\uvicorn\server.py", line 65, in run
return asyncio.run(self.serve(sockets=sockets))
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\asyncio\runners.py", line 194, in run
return runner.run(main)
^^^^^^^^^^^^^^^^
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\asyncio\runners.py", line 123, in run
raise KeyboardInterrupt()
KeyboardInterrupt
INFO: Started server process [18300]
INFO: Waiting for application startup.
INFO: Application startup complete.
3. SwaggerUI 및 화면 확인
1) Swagger 문서 확인
브라우저에 http://127.0.0.1:8000/docs 입력 시
POST API가 추가된 것 확인
POST 클릭 시 request body 도 예시와 함께 문서에 반영되어 있는 것 확인
다만, Responses에 대해서는 typing을 적용 안 해줬기에 아직 정상적으로 적용되지 않음 (추후에 바꿔볼 것임)
2) Swagger 문서에서 POST API 호출 실습
Try it out 클릭
해당 딕셔너리에
"id" 값에 4 넣어주기
"contents" 값에 "실전! FastAPI 섹션 3 수강" 넣어주기
"is_done" 값에 false 넣어주기
Execute 클릭 시 todos path로 POST 요청이 간 것 확인
Curl 내용을 보면 안에 있는 내용으로 request body가 전달된 것 확인
요청에 대한 응답:
우리가 새로 생성한 todo가 위와 같은 식으로 응답으로 오는 것 확인
이때까지의 코드들:
# .../todos/src/main.py 내용
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
# 첫 화면 API
@app.get("/")
def health_check_handler():
return {"ping": "pong"}
# POST 생성을 위해 사용자로부터 전달받을 request 클래스 생성
class CreateToDoRequest(BaseModel):
id: int
contents: str
is_done: bool
# 데이터베이스 역할하는 딕셔너리 생성
todo_data = {
1: {
"id": 1,
"content": "실전! FastAPI 섹션 0 수강",
"is_done": True,
},
2: {
"id": 2,
"content": "실전! FastAPI 섹션 1 수강",
"is_done": False,
},
3: {
"id": 3,
"content": "실전! FastAPI 섹션 2 수강",
"is_done": False,
},
}
# GET Method 사용하여 전체 조회 API
@app.get("/todos")
def get_todos_handler(order: str | None = None):
rt = list(todo_data.values())
if order and order == "DESC":
return rt[::-1]
return rt
# GET Method 사용하여 단일 조회 API
@app.get("/todos/{todo_id}")
def get_todo_handler(todo_id: int):
return todo_data.get(todo_id, {})
# POST Method 사용하여 todo 생성 API
@app.post("/todos")
def create_todos_handler(request: CreateToDoRequest):
todo_data[request.id] = request.dict()
return todo_data[request.id]
