FastAPI 의 jsonable_encoder 들여다보기

jsonable_encoder 이란?

FastAPI 에서 제공하는 인코더 함수로, 보통은 클라이언트로 전송하기 전에 응답하는 객체를 json 으로 인코딩할 수 있도록 변환해 주는 역할을 합니다. (참고로 json 으로 변환해 주는 것은 아닙니다!)

 

실제로 FastAPI 는 jsonable_encoder 를 어떻게 사용할까요? 아래 코드는 요청이 유효하지 않을 때 에러를 응답하는 함수입니다.

# fastapi.exception_handlers.py

async def request_validation_exception_handler(
    request: Request, exc: RequestValidationError
) -> JSONResponse:
    return JSONResponse(
        status_code=HTTP_422_UNPROCESSABLE_ENTITY,
        content={"detail": jsonable_encoder(exc.errors())},
    )

 

exc.errors() 과정에서 혹시 json 으로 변환할 수 없는 타입이 있다면 이를 jsonable 하게 변환(직렬화)해주죠.

그렇다면 만약 jsonable_encoder 를 사용하지 않는다면 어떻게 해야 할까요? 아니, 저처럼 그 존재를 모르고 있었다면 🫠!?🫠!?🫠!?

 

json 으로 변환할 수 없는 Pydantic BaseModel 이나 Datetime 객체 등을 일일히 직접 jsonable 한 형태로 바꾸어야겠죠... 그런데, 언제 그런거 신경쓰며 개발하겠습니까! 그러니 우리는 jsonable_encoder 를 적재적소 활용해봅시다!

 

이제 뒤로가기를 하셔도 됩니다! 하. 지. 만! 혹시나 jsonable_encoder 가 어떻게 Python 의 모든 타입들을 jsonable 하게 변환해줄 수 있는지 궁금하신 분들이 계시다면! 이 글을 끝까지 봐주시기 바랍니다.🙏

 

자 그럼 이제부터 jsonable_encoder 함수를 파헤쳐보겠습니다. ⛏️

 

---

 

 

jsonable_encoder 파라미터

def jsonable_encoder(
    obj: Any,
    include: Optional[IncEx] = None,
    exclude: Optional[IncEx] = None,
    by_alias: bool = True,
    exclude_unset: bool = False,
    exclude_defaults: bool = False,
    exclude_none: bool = False,
    custom_encoder: Optional[Dict[Any, Callable[[Any], Any]]] = None,
    sqlalchemy_safe: bool = True,
) -> Any:

 

이 코드는 jsonable_encoder 의 파라미터 부분입니다. FastAPI 답게 타입이 잘 명시된게 보이네요.

 

각 파라미터를 설명하면 다음과 같습니다. 

FastAPI 공식문서 캡처해버리기

 

파라미터 중에 obj 는 실제 우리가 jsonable 한 값을 얻기 위한 넣는 객체이고, 나머지는 특정 필드를 포함하거나, 제외하거나, 기본값을 제외하거나, None 을 제외하거나 등등의 옵션 인자입니다.

 

그렇다면 내부 구현은 어떻게 이루어져있을까요?

 

---

 

 

jsonable_encoder 내부

 

다시 한번 말하자면 jsonable_encoder 는 Python에서 사용하는 모든 객체를 핸들링하며 jsonable 한 객체로 반환합니다. 그 과정을 구현된 코드 순서대로 살펴보겠습니다.

 

 

1. 커스텀 인코더 설정

custom_encoder = custom_encoder or {}

 

여기서 custom_encoder 매개변수는 기본값이 None입니다. 이 줄은 custom_encoder가 None인 경우 빈 dict 로 설정합니다. 즉, 사용자가 커스텀 인코더를 제공하지 않으면, 빈 dict 를 사용하게 됩니다.

 

 

2. 사용자 정의 인코더 적용

if custom_encoder:
    if type(obj) in custom_encoder:
        return custom_encoder[type(obj)](obj)
    else:
        for encoder_type, encoder_instance in custom_encoder.items():
            if isinstance(obj, encoder_type):
                return encoder_instance(obj)

이 부분은 사용자가 제공한 커스텀 인코더가 있는 경우, 해당 인코더를 사용하여 객체를 직렬화합니다. obj의 타입이 custom_encoder에 정의된 타입 중 하나와 일치하면, 해당 인코더 함수를 사용해 obj를 변환합니다.

 

 

3. 포함 및 제외 설정

if include is not None and not isinstance(include, (set, dict)):
    include = set(include)
if exclude is not None and not isinstance(exclude, (set, dict)):
    exclude = set(exclude)

include와 exclude 매개변수는 객체의 특정 부분을 포함하거나 제외하기 위해 사용됩니다. 이 코드는 include와 exclude가 set 이나 dict 이 아닌 경우 set 으로 변환합니다.

 

 

4. Pydantic 모델 처리

if isinstance(obj, BaseModel):
    obj_dict = _model_dump(...)
    return jsonable_encoder(obj_dict, ...)

obj가 Pydantic BaseModel의 인스턴스인 경우, 이를 dict로 변환한 다음 jsonable_encoder 함수에 다시 전달합니다. 참고로 _model_dump 는 BaseModel 을 dict 로 변환하는 메서드이고 mode="json" 옵션까지 추가하여 jsonable 한 dict 로 변환하도록 하네요.

 

 

5. 데이터 클래스 처리

if dataclasses.is_dataclass(obj):
    obj_dict = dataclasses.asdict(obj)
    return jsonable_encoder(obj_dict, ...)

obj가 데이터 클래스인 경우, 이를 dict로 변환한다음, 마찬가지로 jsonable_encoder 함수에 다시 전달합니다. 이렇듯 자신을 재귀적으로 호출하는 이유는 직렬화 로직을 최대한 재활용하려는 것으로 보이네요.

 

 

6. 열거형(Enum) 처리

if isinstance(obj, Enum):
    return obj.value

obj가 열거형(Enum)인 경우, value 값을 그대로 반환합니다.

 

 

7. 파일 경로 처리

if isinstance(obj, PurePath):
    return str(obj)

obj가 파일 경로(PurePath)인 경우, 이를 문자열로 변환하여 반환합니다.

 

 

8. 기본 데이터 타입 처리

if isinstance(obj, (str, int, float, type(None))):
    return obj

obj가 문자열, 정수, 부동소수점 또는 None 타입인 경우, 그대로 반환합니다.

 

 

9. dict 처리

if isinstance(obj, dict):
     encoded_dict = {}
     allowed_keys = set(obj.keys())
     if include is not None:
         allowed_keys &= set(include)
     if exclude is not None:
         allowed_keys -= set(exclude)
     for key, value in obj.items():
         if (
             (
                 not sqlalchemy_safe
                 or (not isinstance(key, str))
                 or (not key.startswith("_sa"))
             )
             and (value is not None or not exclude_none)
             and key in allowed_keys
         ):
             encoded_key = jsonable_encoder(key, ...)
             encoded_value = jsonable_encoder(value, ...)
             encoded_dict[encoded_key] = encoded_value
     return encoded_dict

obj가 dict인 경우, 각 키와 값을 jsonable_encoder를 사용하여 json 으로 직렬화 가능한 형태로 변환합니다. include 와 exclude 가 이 과정에서 사용되는 것을 볼 수 있습니다.

 

 

10. 컬렉션 처리

if isinstance(obj, (list, set, frozenset, GeneratorType, tuple, deque)):
    encoded_list = []
    for item in obj:
        encoded_list.append(jsonable_encoder(item, ...))
    return encoded_list

obj가 리스트, 세트, 튜플 등의 컬렉션인 경우, 각 개별 요소를 jsonable_encoder를 사용하여 변환합니다.

 

 

11. 등록된 인코더 사용

if type(obj) in ENCODERS_BY_TYPE:
    return ENCODERS_BY_TYPE[type(obj)](obj)
for encoder, classes_tuple in encoders_by_class_tuples.items():
    if isinstance(obj, classes_tuple):
        return encoder(obj)

이 부분은 obj의 타입에 대해 등록된 인코더가 있는 경우 해당 인코더를 사용합니다.

 

 

12. 예외 처리

try:
    data = dict(obj)
except Exception as e:
    try:
        data = vars(obj)
    except Exception as e:
        raise ValueError(errors) from e
return jsonable_encoder(data, ...)

여기서는 obj를 dict 으로 변환하려고 시도하는데요. 만약 실패한다면 객체의 변수들을 사용하여 dict을 생성합니다. 그리고 이 dict을 jsonable_encoder에 다시 전달하여 직렬화를 시도합니다.

 

 

후우~ 함수 내부에 대한 설명은 여기까지 하겠습니다. 생각보다 굉장히 많은 일을 하는 함수였네요!

 

 

---

 

 

마무리

이제 여러분은 jsonable_encoder의 내부 동작까지 알게 되었습니다! 👏👏👏 

 

자, 그렇다면 앞으로 jsonable_encoder를 언제 사용하면 될까요?

 

• 복잡한 데이터 변환: 여러분이 다루는 데이터가 json으로 변환하기 어렵다면, jsonable_encoder를 사용해보세요. 복잡한 데이터를 간단하게 변환해줍니다.
• 에러 핸들링에 사용: jsonable_encoder를 사용하여 다양한 예외를 안전하게 응답할 수 있습니다. 아주 유용하죠. (사실 제가 이거 때문에 알게되었거든요 😉)
• 커스텀 처리 필요 시: 특정 데이터 타입에 대해 특별한 직렬화 방법이 필요하다면, custom_encoder 매개변수를 사용해보세요. 맞춤형 직렬화를 구현하실 수 있습니다. 어떻게 커스텀할 수 있는지 짧은 예시 코드를 보여드릴게요!

from datetime import datetime
from fastapi.encoders import jsonable_encoder


# 사용자 정의 클래스
class 그냥데이터:
    def __init__(self, data, timestamp):
        self.data = data
        self.timestamp = timestamp


# 맞춤형 직렬화 함수
def 안녕잘가함수(obj):
    return {"안녕": obj.data, "잘가": obj.timestamp.isoformat()}


# 사용자 정의 객체 생성
데이터 = 그냥데이터("Some data", datetime.now())

# jsonable_encoder에 맞춤형 함수 전달
encoded_data = jsonable_encoder(데이터, custom_encoder={그냥데이터: 안녕잘가함수})

print(encoded_data)  # {'안녕': 'Some data', '잘가': '2024-01-18T21:28:03.711478'}

 

제 입맛대로 결과물이 출력 되는게 보이시나요? ㅎㅎ

후우.. 마지막까지 알찼다.. 그쵸?

 

이 글을 통해 여러분도 jsonable_encoder를 유용하게 활용하시기 바랍니다! 😊

 

 

---

 

 

공식문서 & 소스코드

Encoders - jsonable_encoder - FastAPI

 

https://github.dev/tiangolo/fastapi/blob/d74b3b25659b42233a669f032529880de8bd6c2d/fastapi/encoders.py#L102

 

 

 

ps. 앞서 jsonable_encoder 이 재귀로 동작한다고 말씀드렸는데요. 이와 관련해 성능 이슈에 대한 글이 있어 링크를 남깁니다.

[FastAPI] jsonable_encoder에 의한 Reponse Model Validation 성능이슈