覆盖 FastAPI 中jsonable_encoder的默认编码器



我有使用Fast API的代码看起来像这样:

class EnumTestT(Enum):
test_t_value = 0
object = { 
test: test_t_value
}
enum_mapping = {
test_t_value: "Test"
}
def enum_encoder(val: EnumTestT) -> str:
return enum_mapping[val]
custom_encoder = {
EnumTestT: enum_encoder
}
@app.get("/test")
async def test_get():
return jsonable_encoder(object, custom_encoder=custom_encoder)

问题是jsonable_encoder在默认值之后应用自定义编码器。 有没有办法在默认编码器之前应用它们。 因为对于Enum和任何派生类,将报告枚举的值而不是映射的值。

现在我在json.dumps中使用自定义编码器,如下所示:

class TestEncoder(JSONEncoder):
def default(self, obj):
if isinstance(obj, EnumTestT):
return TestEncoder.process_enum_test_t(obj)

FastAPI 应用中的响应将是:

json_str = json.dumps(json_list, cls=TestEncoder).encode('utf-8')
return Response(media_type="application/json", content=json_str)

使用自定义编码器的 FastAPI 的问题在于,在调用所有标准编码器后调用自定义编码器,并且无法覆盖该顺序。

FastAPI 使用ENCODERS_BY_TYPE(来自 pydantic.json(对一些基本数据类型进行编码。

ENCODERS_BY_TYPE: Dict[Type[Any], Callable[[Any], Any]] = {
bytes: lambda o: o.decode(),
Color: str,
datetime.date: isoformat,
datetime.datetime: isoformat,
datetime.time: isoformat,
datetime.timedelta: lambda td: td.total_seconds(),
Decimal: decimal_encoder,
Enum: lambda o: o.value,

所以对我来说覆盖默认的日期时间编码,就像

ENCODERS_BY_TYPE[datetime] = lambda date_obj: date_obj.strftime("%Y-%m-%d %H:%M:%S")

灵感来自 Coder vx

import datetime
from fastapi.responses import JSONResponse 
...
content = something to serialize 
JSONResponse(jsonable_encoder(content,
custom_encoder={datetime.datetime: lambda date_obj: 
date_obj.strftime("%Y-%m-%d %H:%M:%S")})))

更进一步,您可以使用自定义编码器对 FastAPI/Starlette 的JSONResponse进行子类化(请参阅JSONResponse的实现(:

import json
from json import JSONEncoder
from fastapi.responses import JSONResponse

class MyJSONEncoder(JSONEncoder):
def default(self, obj):
if isinstance(obj, UUID):
return str(obj)
elif isinstance(obj, datetime):
return obj.strftime('%Y-%m-%dT%H:%M:%SZ')
else:
return super().default(obj)

class MyJSONResponse(JSONResponse):
def render(self, content: Any) -> bytes:
return json.dumps(
content,
ensure_ascii=False,
cls=MyJSONEncoder,
allow_nan=False,
indent=None,
separators=(",", ":"),
).encode("utf-8")

然后在您的路线中,您可以返回MyJSONResponse(some_dict_obj).注意:json不支持将 UUID 作为密钥,因此您需要找到一种方法将它们转换为intstr

您可以通过返回如下所示的fastapi.responses.Response来使用任何序列化程序:

from fastapi.responses import Response
return Response(
content=CustomJsonEncoder().to_string(result),
media_type='application/json'
)

通过编写继承自json.JSONEncoder的类,可以重写方法def default(self, obj)以更改特定数据类型的序列化。

class CustomJsonEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, specific_class_type):
return specific_serialization(obj)
pass

最新更新