앞 단계 참조 링크:
- 상태 코드: 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
00 개요
- 앞서 swagger UI를 통해 실습 진행했을 때 데이터베이스에서 조회 한 이 ORM 객체를 그대로 리턴을 해주어도 FastAPI에 의해 이 객체가 JSON 형태로 변환이 되어서 잘 응답한 것을 확인했음
- 일반적으로는, 데이터를 직접 리턴하지 않고 데이터가 표현되는 부분을 따로 분리해서 처리함
- 목적: 반환할 Response를 따로 정의한 API 생성하는 실습 진행
01 HTTP Response 처리
1. response.py 파일 생성
- get_todos API가 어떤 식으로 응답해야 하는지, 즉 schema를 정의해 주기 위한 response.py 파일 생성
- 경로: src/schema/response.py
# mkdir /c/Users/관리자/Desktop/projects/todos/src/schema
# touch /c/Users/관리자/Desktop/projects/todos/src/schema/response.py
2. 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:
orm_mode = True
class ListToDoSchema(BaseModel):
todos: List[ToDoSchema]
- from pydantic import BaseModel
- pydantic에서 BaseModel 참조
- from typing import List
- typing에서 List 참조 (타입힌트 참조: 2024.05.08 - [Python/기본문법] - 파이썬 Type Hint (타입 정보 명시적 표시))
- class ToDoSchema(BaseModel):
- ToDoSchema라는 이름으로 BaseModel을 상속 받아서 클래스 생성
- ToDo 데이터랑 동일하게 typing 적용하기
- id: int
- contents: str
- is_done: bool
- pydantic의 sqlalchemy를 바로 읽어줄 수 있도록 하려면 옵션 하나를 추가해줘야 함
- class Config:
- Config이라는 클래스 정의해 준 다음
- orm_mode = True
- 이렇게 적어주면 pydantic에 정의한 orm_mode하는 걸 사용할 수 있게 됨
- → 그래서 ToDoSchema에 sqlalchemy orm 객체를 던져주게 되면 이 pydantic이 그 orm 객체를 잘 해석해서 class ToDoSchema에 맞게 변경을 시켜줌
- class Config:
- class ListToDoResponse(BaseModel):
- ListToDoResponse라는 이름으로 BaseModel을 상속 받아서 클래스 생성
- todos: List[ToDoSchema]
- todos라는 키에 ToDoSchema를 리스트에 담아 typing 적용하기
- 이 ListToDoResponse라는 클래스를 응답에 활용할 것임
- 그 안에 todos라는 키로 ToDo 전체 데이터를 리스트 형태로 담아서 리턴하게 될 것
- ToDoSchema가 sqlalchemy의 orm 객체를 전달받아 우리가 정의한 형태에 맞게 데이터를 변환하고 최종적으로 리턴해주게 됨
- 이미 우리가 orm.py에 생성한 컬럼과 여기 ToDoSchema에 표현되는 컬럼들이 동일함
- ☆이렇게 Reponse를 분리하고 정의하는 이유
- 지금은 컬럼과 응갑의 구조가 단순하지만 구조가 복잡해질 경우 (e.g., 컬럼 간의 연산이 있거나 혹은 이 객체를 뭔가 중첩된 구조로 반환을 한다거나 컨텐츠 값을 제외하고 리턴을 하는 등)
- 다양한 use case가 있을 수 있기 때문에 미리 이렇게 response 객체 분리 시 유연하게 코드 변경 가능하도록 하기 위함
3. 실습으로 확인 - python console에서
0) 실습 환경 준비
- 가상환경 활성화, docker 컨테이너 시작, 경로 src로 이동
$ . /c/Users/관리자/Desktop/projects/todos/Scripts/activate # 가상환경 활성화
(todos)$ docker ps -a # docker 컨테이너 목록 및 상태 확인
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
0c3283e97292 mysql:8.0 "docker-entrypoint.s…" 2 weeks ago Exited (0) 24 hours ago todos
2597a124c27d docker/welcome-to-docker:latest "/docker-entrypoint.…" 2 weeks ago Exited (0) 2 weeks ago welcome-to-docker
(todos)$ docker start todos # todos 컨테이너 시작하기
todos
(todos)$ cd /c/Users/관리자/Desktop/projects/todos/src # python console을 시작할 src 경로로 이동
1) python console 열기
(todos)$ 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.
>>>
2) 다음을 입력
>>> from schema.response import ToDoSchema
>>> from database.orm import ToDo
>>> todo = ToDo(id=100, contents="text", is_done=True)
>>> ToDoSchema.from_orm(todo)
- from schema.response import ToDoSchema
- schema 폴더 안에 있는 response 모듈에서 우리가 생성한 ToDoSchema 참조
- from database.orm import ToDo
- database 폴더 안에 있는 orm 모듈에서 ToDo 모델 참조
- todo = ToDo(id=100, contents="text", is_done=True)
- todo 객체 생성하기→ ToDo 모델 참조하여 객체 생성함, 이 객체는 데이터베이스에 저장되지 않고 그냥 python console에서만 띄워지는 객체임
- ToDoSchema.from_orm(todo)
- ToDoSchema에서 from_orm이라는 메소드를 이용하여 우리가 생성한 todo 객체를 여기에 전달해주면
ToDoSchema 객체 생성되야 됨
- ToDoSchema에서 from_orm이라는 메소드를 이용하여 우리가 생성한 todo 객체를 여기에 전달해주면
- 그런데 나는 에러남
>>> ToDoSchema.from_orm(todo)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "C:\Users\관리자\AppData\Local\Programs\Python\Python312\Lib\site-packages\pydantic\main.py", line 1171, in from_orm
raise PydanticUserError(
pydantic.errors.PydanticUserError: You must set the config attribute `from_attributes=True` to use from_orm
- 이유1: 'class Config: orm_mode = True'를 잘못된 줄에 작성했었음
- 이유2: 내가 다운받은 pydantic 버전 (version 2)이 강사님이 사용하시는 pydantic 버전 (version 1.10.15)이랑 다름
- pip uninstall pydantic 후 version1으로 다시 다운받아야 됨
$ pip uninstall pydantic
$ pip install "pydantic<2.0"
- 문제 해결 후 다시 실행 결과
>>> from schema.response import ToDoSchema
>>> from database.orm import ToDo
>>> todo = ToDo(id=100, contents="text", is_done=True)
>>> ToDoSchema.from_orm(todo)
ToDoSchema(id=100, contents='text', is_done=True)
- ToDoSchema 객체 생성된 것을 확인: ToDoSchema(id=100, contents='text', is_done=True)
- 그래서 별도로 작업을 해주지 않아도 앞서 orm_mode를 True로 줬기 때문에 (response.py에서 class Config에 정의됨)
이 from_orm을 이용 시 우리가 파이썬 콘솔에서 생성한 todo 객체를 이런식으로 pydantic schema로 쉽게 변환 가능 (ToDoSchema(id=100, contents'test', is_done=True))
4. main.py에 쓰기
- 목적: from_orm을 API에 적용하기
- API를 리턴하는 부분에서 todo를 직접 리턴해줄 것이 아니라 ListToDoResponse()로 해주게 될 것
# /c/Users/관리자/Desktop/projects/todos/src/main.py 내용
from fastapi import FastAPI, Body, HTTPException, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
from typing import List
from database.connection import get_db
from database.repository import get_todos
from database.orm import ToDo
from schema.response import ToDoSchema, ListToDoResponse # 추가
...
# GET Method 사용하여 전체 조회 API - DB 참조
@app.get("/todos", status_code=200)
def get_todos_handler(
order: str | None = None,
session: Session = Depends(get_db),
) -> ListToDoResponse: # 추가
todos: List[ToDo] = get_todos(session=session)
if order and order == "DESC":
# return todos[::-1] # 삭제
return ListToDoResponse( # 추가
todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
)
# return todos # 삭제
return ListToDoResponse( # 추가
todos=[ToDoSchema.from_orm(todo) for todo in todos]
)
- from schema.response import ToDoSchema, ListToDoResponse
- 이번에 생성한 ToDoSchema, ListToDoResponse 참조
- return ListToDoResponse(
- todo를 직접 리턴해줄 것이 아니라 ListToDoResponse()로 리턴을 해주기
- ) -> ListToDoResponse:
- Response 타입을 ListToDoResponse로 지정 (타입 힌트)
- todos= [ToDoSchema.from_orm(todo) for todo intodos[::-1]] → 역정렬
- todos = [ToDoSchema.from_orm(todo) for todo in todos]
- ListToDoResponse()에 todo 전달해주는 것
- response.py에 보면 (todos: List[ToDoSchema])와 같이 리스트에 ToDoSchema가 담긴 형태로 전달해줘야 됨
- 이 typing을 맞춰주기 위해 list comprehension을 사용할 것
- list comprehension: 대괄호 "[", "]"로 감싸고 내부에 for문과 if 문을 사용하여 반복하며 조건에 만족하는 것만 리스트로 생성하는 문법
- todos를 순회하면서 todo 하나를 from_orm에 전달해줘서 ToDoSchema 객체가 계속 생성이 되며 todos변수에 리스트 형태로 담기게 됨
- todos= [ToDoSchema.from_orm(todo) for todo intodos[::-1]] → 역정렬
- 그래서 우리가 앞서 생성했던 todos 데이터를 이 전부 ToDoSchema로 바꿔서 리턴을 해주는 것
5. 실습으로 확인 - SwaggerUI 문서에서
0) 실습 환경 준비
- 가상환경 활성화, docker 컨테이너 시작, 경로 src로 이동, uvicorn 실행
# 가상환경 활성화
$ . /c/Users/관리자/Desktop/projects/todos/Scripts/activate
# docker 컨테이너 목록 및 상태 확인
(todos)$ docker ps -a
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
0c3283e97292 mysql:8.0 "docker-entrypoint.s…" 2 weeks ago Exited (0) 24 hours ago todos
2597a124c27d docker/welcome-to-docker:latest "/docker-entrypoint.…" 2 weeks ago Exited (0) 2 weeks ago welcome-to-docker
# todos 컨테이너 시작하기
(todos)$ docker start todos
todos
# src 경로로 이동
(todos)$ cd /c/Users/관리자/Desktop/projects/todos/src
# uvicorn 실행
(todos)$ 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 [8444] using WatchFiles
INFO: Started server process [6332]
INFO: Waiting for application startup.
INFO: Application startup complete.
1) 브라우저에 SwaggerUI 문서 띄우기
- http://127.0.0.1:8000/docs 입력
- GET /todos Get Todos Handler 클릭 > Responses에 Code: 200에 Example Value | Schema 중 Example Value 클릭 시
- Schema 클릭 시
- array 형태로 todo 아이템 하나하나 리턴되는 방식으로 바뀐 것 확인
- Schema 의 ListToDoResponse 안 Expand all 클릭 시
- Try it out > Execute
- 앞서 생성한 3개의 todo가 우리가 변경한 스키마에 맞게 리턴되는 것 확인
이때까지의 코드들:
- /c/Users/관리자/Desktop/projects/todos/src/main.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/response.py
# /c/Users/관리자/Desktop/projects/todos/src/main.py 내용
from fastapi import FastAPI, Body, HTTPException, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
from typing import List
from database.connection import get_db
from database.repository import get_todos
from database.orm import ToDo
from schema.response import ToDoSchema, ListToDoResponse
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", status_code=200)
# 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 - DB 참조
@app.get("/todos", status_code=200)
def get_todos_handler(
order: str | None = None,
session: Session = Depends(get_db),
) -> ListToDoResponse:
todos: List[ToDo] = get_todos(session=session)
if order and order == "DESC":
# return todos[::-1]
return ListToDoResponse(
todos=[ToDoSchema.from_orm(todo) for todo in todos[::-1]]
)
# return todos
return ListToDoResponse(
todos=[ToDoSchema.from_orm(todo) for todo in todos]
)
# GET Method 사용하여 단일 조회 API
@app.get("/todos/{todo_id}", status_code=200)
def get_todo_handler(todo_id: int):
todo = todo_data.get(todo_id)
if todo:
return todo
raise HTTPException(status_code=404, detail="ToD Not Found")
# POST Medthod 사용하여 todo 생성 API
@app.post("/todos", status_code=201)
def create_todos_handler(request: CreateToDoRequest):
todo_data[request.id] = request.dict()
return todo_data[request.id]
# PATCH Method 사용하여 is_done 값 수정 API
@app.patch("/todos/{todo_id}", status_code=200)
def update_todo_handler(
todo_id: int,
is_done: bool = Body(..., embed=True)
):
todo = todo_data.get(todo_id)
if todo:
todo["is_done"] = is_done
return todo
raise HTTPException(status_code=404, detail="ToDo Not Found")
# DELETE Method 사용하여 todo 아이템 삭제 API
@app.delete("/todos/{todo_id}", status_code=204)
def delete_todo_handler(todo_id: int):
todo = todo_data.pop(todo_id, None)
if todo:
return
raise HTTPException(status_code=404, detail="ToDo Not Found")
# /c/Users/관리자/Desktop/projects/todos/src/database/orm.py 내용
from sqlalchemy import Boolean, Column, Integer, String
from sqlalchemy.orm import declarative_base
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)
def __repr__(self):
return f"ToDo(id={self.id}, contents={self.contents}, is_done={self.is_done})"
# /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/repository.py 내용
from sqlalchemy import select
from sqlalchemy.orm import Session
from typing import List
from database.orm import ToDo
def get_todos(session: Session) -> List[ToDo]:
return list(session.scalars(select(ToDo)))
# /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:
orm_mode = True
class ListToDoResponse(BaseModel):
todos: List[ToDoSchema]
'Web 개발 > FAST API (인프런 강의 내용)' 카테고리의 다른 글
2 실습8 ORM 적용 - Refactoring (0) | 2024.06.01 |
---|---|
2 실습7 ORM 적용 - GET 단일조회 API (0) | 2024.05.14 |
2 실습5 ORM 적용 - GET 전체조회 API (0) | 2024.05.03 |
2 실습4 ORM 모델링 (0) | 2024.05.02 |
2 실습3 데이터베이스 연결 (0) | 2024.04.25 |