[FastAPI] Pydantic 2.0 적용하기
2023년 6월 30일, Pydantic 2.0 버전이 정식으로 출시되었다.
Pydantic이란?
Pydantic은 Python에서 사용할 수 있는 데이터 검증 라이브러리이다. 최근 들어 Python 웹개발을 포함한 여러 Python 개발 분야에서 Type Annotation을 적극적으로 활용하여 개발하는 것이 트랜드가 되었다.
기존의 Python개발 트랜드가 Type Hint를 명시하지 않았던 것은 아니지만 Java를 비롯한 다른 언어와는 다르게 Python에서는 명시적으로 메서드 인수나 메서드 리턴 타입을 명시해놓을 수는 없었다.
Python 3.5버전 이후, Python 기본 라이브러리에 'typing'라이브러리가 추가가 되면서 메서드나 변수의 타입을 명시해놓을 수 있게 되었다.
typing 라이브러리가 추가 되기 전 type hint는 다음과 같이 메서드 내부에 주석으로 표시해놓는 것이 트랜드였다.
num = 1 # type: int
def duplicate(string, number = 2):
# type: (str, int) -> list
return [string] * number
typing 라이브러리를 사용하면 다음과 같이 메서드의 type annotation을 적용할 수 있다.
num: int = 1
def duplicate(string: str, number: int = 2) -> list:
return [string] * number
이렇게 Python에 type annotation을 적용할 수 있게 되면서 이 type annotation을 좀 더 강력하게 활용할 수 있게 나온 라이브러리가 바로 Pydantic이다.
웹 개발을 하면 다양한 표준으로 넘어오는 데이터들을 Serializing, 유효성 검증에 대한 로직을 필수로 구현해야하는데, 이 Pydantic이 제공하는 표준을 적용하면 JSON 스키마 생성, 유효성 검증에 대한 로직을 직접 구현하지 않아도 안전하고 강력한 Data Serializing을 우리 웹 서버에 적용할 수 있다.
FastAPI는 이 Pydantic 기능을 적극적으로 도입하여 사용하는 것을 공식 문서에서부터 추천하고 있다.
그리고, Pydantic이 2.0버전으로 업데이트 되면서 Pydantic 로직의 일부가 변경되고 deprecated됨에 따라 FastAPI에서도 Pydantic을 적용하여 스키마 로직을 짜는데 약간의 변화가 생겼는데, 필자가 직접 FastAPI 서버 로직을 짜면서 겪었던 로직 구현의 변화에 대해 서술해보겠다.
Settings
Django는 프로젝트 init을 하면 자동으로 생성되는 파일 중에 settings.py라는 파일을 자동으로 생성해준다. Django 프로젝트의 여러 셋업들은 이 모듈을 바탕으로 돌아가기 때문에 웹 서버 셋업을 모두 하나의 모듈에서 할 수 있다는 장점이 있다.
그러나 FastAPI는 이러한 프레임워크에서 프로젝트 셋업을 강제하지 않는다. 따라서 FastAPI로 서버를 짜는 개발자는 자기 취향대로 프로젝트 셋업 로직을 짤 수 있다는 장점이자 단점이 존재한다.
이에 나는 Pydantic을 사용해서 프로젝트 세팅값을 설정해두는 편이다. Pydantic은 1.X버전부터 'BaseSettings'라는 클래스로 Settings 관련 필드들을 스키마로 짤 수 있는 기능이 존재했다.
이걸 활용한 FastAPI 프로젝트 셋업 스키마는 다음과 같이 구현할 수 있다.
from typing import Any, Dict, Optional
from pydantic import AnyUrl, BaseSettings, Field
class AppSettings(BaseSettings):
LOGGING_DEBUG_LEVEL: bool = Field(
default=True,
description="True: DEBUG mode, False:: INFO mode",
)
DEBUG_ALLOW_CORS_ALL_ORIGIN: bool = Field(
default=True,
description="If True, allow origins for CORS requests.",
)
DEBUG_ALLOW_NON_CERTIFICATED_USER_GET_TOKEN: bool = Field(
default=True,
description="If True, allow non-cerficiated users to get ESP token.",
)
THREAD_POOL_SIZE: Optional[int] = Field(
default=10,
description="Change the server's thread pool size to handle non-async function",
)
SECRET_KEY: str = Field(
default="some_secret_key",
description="Secret key to be used for issuing HMAC tokens.",
)
DATABASE_URI: AnyUrl = Field(
default="postgresql+asyncpg://<user>:<password>@<host>:<port>/<db_name>",
description="PosstgreSQL connection URI.",
)
DATABASE_OPTIONS: Dict[str, Any] = Field(
default={
"connect_args": {
"keepalives": 1,
"keepalives_idle": 30,
"keepalives_interval": 15,
},
"pool_pre_ping": True,
"pool_recycle": 15 * 60,
"pool_size": 50,
"max_overflow": 50,
"pool_use_lifo": True,
},
description="PosstgreSQL option to create a connection.",
)
class Config:
env_file = ".env"
2.0 버전으로 넘어오면서 이 셋업 스키마를 구축하는 문법에도 변화가 생겼다.
먼저 Config class 선언에 대한 변화다.
Pydantic 2.0 버전에서도 위와 같이 그대로 셋업 스키마를 구축하면 Config class는 deprecated 되었다는 warning이 뜰 것이다. 로그 내역을 보게 되면 Config class 말고 ConfigDict를 사용하라는 warning이 뜨는데, 이 warning을 따라서 Config 이라는 class이름을 단순히 ConfigDict로 변경하면 특정 상황에서 셋업 스키마에 정의했던 필드가 프로젝트 런타임에 제대로 적용이 안되는 상황이 생긴다.
이에 대한 대안으로 나온 것이 pydantic-settings 이라는 라이브러리이다.
다음 명령어로 pydantic-settings 라이브러리를 설치할 수 있다.
pydantic-settings 설치
$ pip install pydantic-settings
설치를 완료하면 위의 셋업 스키마의 Config 클래스 선언부를 SettingsConfigDict를 사용하여 다음과 같이 수정하면 된다.
from typing import Any, Dict, Optional
from pydantic import AnyUrl, Field
from pydantic_settings import BaseSettings, SettingsConfigDict # 추가
... 앞 부분과 동일 ...
DATABASE_OPTIONS: Dict[str, Any] = Field(
default={
"connect_args": {
"keepalives": 1,
"keepalives_idle": 30,
"keepalives_interval": 15,
},
"pool_pre_ping": True,
"pool_recycle": 15 * 60,
"pool_size": 50,
"max_overflow": 50,
"pool_use_lifo": True,
},
description="PosstgreSQL option to create a connection.",
)
model_config = SettingsConfigDict(env_file=".env") # class Config 선언 대신 이렇게 수정.
기존에 pydantic 라이브러리에 있던 BaseSettings대신 pydantic-settings 라이브러리의 BaseSettings을 사용하고 SettingsConfigDict를 import 해서 사용하자.
JSON serializing
웹 요청으로 넘어온 데이터를 JSON으로 직렬화(serializing)하는 과정에서 Request body를 pydantic 스키마로 명시해두었다면, pydantic이 제공해주는 기능을 사용해서 request schema를 JSON으로 직렬화를 할 수 있는데, 2.0 버전에서는 직렬화하는 문법에 변화를 줘야한다.
기존에는 다음과 같이 .dict() 혹은 .parse_obj()를 사용하여 Pydantic 모델을 JSON으로 직렬화하거나 반대로 JSON 데이터를 Pydantic 모델로 파싱하는 것이 가능했다.
from fastapi import FastAPI, HTTPException
from typing import List, Dict
app = FastAPI()
# 가상의 데이터베이스를 모방하는 리스트
fake_db = []
# Pydantic model -> JSON data
@app.post("/posts/", response_model=PostResponse)
def create_post(post_request: PostCreateRequest):
post_data = post_request.dict() # Pydantic 모델을 딕셔너리(JSON)로 변환
post_id = len(fake_db) + 1
# other logic ...
return post_data
# JSON data -> Pydantic model
@app.post("/users/")
def create_user(user_data: Dict):
try:
user = User.parse_obj(user_data) # JSON 데이터를 Pydantic 모델로 변환
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
# other logic ...
return user
Pydantic 2.0 버전에서는 같은 기능을 수행하는 로직을 다음과 같이 구현하면 된다.
from fastapi import FastAPI, HTTPException
from typing import List, Dict
app = FastAPI()
# 가상의 데이터베이스를 모방하는 리스트
fake_db = []
# Pydantic model -> JSON data
@app.post("/posts/", response_model=PostResponse)
def create_post(post_request: PostCreateRequest):
post_data = post_request.model_dump() # Pydantic 모델을 딕셔너리(JSON)로 변환
post_id = len(fake_db) + 1
# other logic ...
return post_data
# JSON data -> Pydantic model
@app.post("/users/")
def create_user(user_data: Dict):
try:
user = User. model_validate(user_data) # JSON 데이터를 Pydantic 모델로 변환
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
# other logic ...
return user
ORM Schema
최신 웹 백엔드 프레임워크는 ORM(Object Relational Mapping)을 활용하여 데이터베이스에 있는 데이터 인스턴스들을 객체로 사용할 수 있도록 기능 구현이 되어 있다.
Django도 마찬가지고 FastAPI도 SQLAlchemy 같은 라이브러리를 활용하면 DB 인스턴스를 객체로 매핑하여 데이터 처리 로직을 구현할 수 있다.
여기에 Pydantic을 적용시키면 Pydantic이 제공해주는 기능인 .from_orm()을 통해 ORM으로 넘어온 객체 인스턴스를 Pydantic 스키마 모델로 변환이 가능하다.
1.X 버전에서는 다음과 같이 로직을 구현하면 됐다.
# 스키마 선언 모듈
from __future__ import annotations
from pydantic import BaseModel, Field
from .common import RoleSchema
class BoardSchema(BaseModel):
id: int = Field(
...,
title="Board's ID (pk)",
description="게시판의 고유 식별자입니다.",
serialization_alias="board_id_pk",
)
name: str = Field(
...,
title="Board's Name",
description="게시판의 이름입니다.",
serialization_alias="board_name",
)
layout: int = Field(
...,
title="Board's Layout",
description="게시판의 레이아웃입니다.",
serialization_alias="board_layout",
)
comment_write_level: RoleSchema = Field(
...,
title="Board's Comment Write Level",
description="게시판의 댓글 작성 권한입니다. (Role pk)",
validation_alias="comment_write_level_role",
serialization_alias="role_role_pk_comment_write_level",
)
class Config:
orm_mode = True
# Router 모듈
...
@app.get("/boards/{board_id}", response_model=BoardSchema)
async def get_board(board_id: int, db: AsyncSession = Depends(get_db)):
query = select(Board).where(Board.id == board_id)
result = await db.execute(query)
board = result.scalar_one_or_none()
if not board:
raise HTTPException(status_code=404, detail="Board not found")
return BoardSchema.from_orm(board) # SQLAlchemy ORM 인스턴스를 Pydantic 모델로 변환
...
BoardSchema에서 class Config 부분에 orm_mode = True로 설정해두면 해당 스키마가 ORM로 넘어온 객체를 읽을 수 있게 된다.
그러나, 앞서 Settings 스키마에서 Config class를 쓰지 말라 했다고 여기서 Config class의 이름을 ConfigDict로 바꿔버리면 이 BoardSchema가 ORM 객체 데이터를 읽지 못한다.
또한 .from_orm()이 2.0 버전에서 deprecated 되었기 때문에 2.0 버전에서는 다음과 같이 로직을 구현하면 된다.
# 스키마 선언 모듈
from __future__ import annotations
from pydantic import BaseModel, Field
from .common import RoleSchema
class BoardSchema(BaseModel):
id: int = Field(
...,
title="Board's ID (pk)",
description="게시판의 고유 식별자입니다.",
serialization_alias="board_id_pk",
)
name: str = Field(
...,
title="Board's Name",
description="게시판의 이름입니다.",
serialization_alias="board_name",
)
layout: int = Field(
...,
title="Board's Layout",
description="게시판의 레이아웃입니다.",
serialization_alias="board_layout",
)
comment_write_level: RoleSchema = Field(
...,
title="Board's Comment Write Level",
description="게시판의 댓글 작성 권한입니다. (Role pk)",
validation_alias="comment_write_level_role",
serialization_alias="role_role_pk_comment_write_level",
)
class Config:
from_attributes = True # orm_mode를 from_attributes로 변경.
# Router 모듈
...
@app.get("/boards/{board_id}", response_model=BoardSchema)
async def get_board(board_id: int, db: AsyncSession = Depends(get_db)):
query = select(Board).where(Board.id == board_id)
result = await db.execute(query)
board = result.scalar_one_or_none()
if not board:
raise HTTPException(status_code=404, detail="Board not found")
return BoardSchema.model_validate(board) # SQLAlchemy ORM 인스턴스를 Pydantic 모델로 변환
...
간단하게 Config class 내부에 선언했던 orm_mode 대신 from_attributes로 선언하고, from_orm 대신 model_validate 메서드를 사용하면 된다.
예시 JSON 스키마 명시 부분
Pydantic으로 스키마를 짜다보면 FastAPI가 자동으로 만들어주는 API문서에 response 예시 같은 값들을 넣기 위해 다음과 같이 Config class에 예시 스키마를 선언해줄 때가 있다.
...
class ExceptionSchema(BaseModel):
timestamp: str = Field(
...,
description="에러가 발생한 시간입니다.",
)
status: int = Field(..., description="에러의 HTTP status code 입니다.")
error: str = Field(
...,
description="에러의 이름입니다.",
)
message: str = Field(
...,
description="에러의 메시지 내용입니다.",
)
errorCode: str = Field(
...,
description="에러의 코드입니다.",
)
path: str = Field(
...,
description="에러가 발생한 경로입니다.",
)
class Config:
json_schema = {
"example": {
"default": {
"timestamp": "2021-10-17T16:55:00.000000Z",
"status": 500,
"error": "INTERNAL_SERVER_ERROR",
"message": "서버 로직에 알 수 없는 오류가 발생했습니다.",
"errorCode": "HB-GENL-000",
"path": "/hub/api/v2/<some/endpoint>",
}
}
}
...
위의 소스코드는 역시 pydantic 1.X 버전의 문법이라서 위를 그대로 적용시켜서 런타임을 돌리면 json_schema 옵션은 deprecated 되었다는 warning이 뜬다.
따라서 2.0 버전 부터는다음과 같이 구현하면 된다.
... 앞 부분과 동일 ...
path: str = Field(
...,
description="에러가 발생한 경로입니다.",
)
class Config:
json_schema_extra = { # json_schema 대신 json_schema_extra 사용
"example": {
"default": {
"timestamp": "2021-10-17T16:55:00.000000Z",
"status": 500,
"error": "INTERNAL_SERVER_ERROR",
"message": "서버 로직에 알 수 없는 오류가 발생했습니다.",
"errorCode": "HB-GENL-000",
"path": "/hub/api/v2/<some/endpoint>",
}
}
}
...
그 외 Pydantic 2.0의 변경사항은 다음 Pydantic 공식 문서를 참고하여 기존 pydantic 로직을 2.0 버전으로 Migration을 하면 된다.
https://docs.pydantic.dev/latest/migration/
그럼 앞서 런타임 로그에서 설레발 쳤던 ConfigDict라는건 뭔가요?
다음 링크를 보면서 참고해보자.
https://docs.pydantic.dev/latest/api/config/#pydantic.config.ConfigDict
마치며
FastAPI는 Django와 다르게 너무 자유로워서 탈이다. 외부 라이브러리의 업데이트에 따라 신경쓸게 생긴다는 것은 어떨때에는 단점으로 느껴지기도..
그래도 FastAPI 사랑하시죠? 네..
????API: 난 X나 빠르다니까?