Update uniform return to custom encoder (#60)

* Add custom jsonable encoder

* Update uniform return to custom encoder

* Add some description of the return structure
This commit is contained in:
Wu Clan
2023-05-20 16:51:22 +08:00
committed by GitHub
parent e43f128086
commit 847d7226f6
3 changed files with 224 additions and 13 deletions

View File

@ -1,19 +1,34 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from datetime import datetime
from typing import Any, Union, Set, Dict
from typing import Any
from fastapi.encoders import jsonable_encoder
from pydantic import validate_arguments, BaseModel
_JsonEncoder = Union[Set[int | str], Dict[int | str, Any]]
from backend.app.utils.encoders import jsonable_encoder
_ExcludeData = set[int | str] | dict[int | str, Any]
__all__ = ['ResponseModel', 'response_base']
class ResponseModel(BaseModel):
"""
统一返回模型, 可在 FastAPI 接口请求中指定 response_model 及更多操作
统一返回模型
.. tip::
如果你不想使用 ResponseBase 中的自定义编码器,可以使用此模型,返回数据将通过 fastapi 内部的编码器直接自动解析并返回
E.g. ::
@router.get('/test', response_model=ResponseModel)
def test():
return ResponseModel(data={'test': 'test'})
@router.get('/test')
def test() -> ResponseModel:
return ResponseModel(data={'test': 'test'})
"""
code: int = 200
@ -25,13 +40,30 @@ class ResponseModel(BaseModel):
class ResponseBase:
"""
统一返回方法
.. tip::
此类中的返回方法将通过自定义编码器预解析,然后由 fastapi 内部的编码器再次处理并返回,可能存在性能损耗,取决于个人喜好
E.g. ::
@router.get('/test')
def test():
return response_base.success(data={'test': 'test'})
"""
@staticmethod
def __encode_json(data: Any):
return jsonable_encoder(data, custom_encoder={datetime: lambda x: x.strftime('%Y-%m-%d %H:%M:%S')})
def __json_encoder(data: Any, exclude: _ExcludeData | None = None, **kwargs):
custom_encoder = {datetime: lambda x: x.strftime('%Y-%m-%d %H:%M:%S')}
kwargs.update({'custom_encoder': custom_encoder})
return jsonable_encoder(data, exclude=exclude, **kwargs)
@staticmethod
@validate_arguments
def success(*, code: int = 200, msg: str = 'Success', data: Any | None = None, exclude: _JsonEncoder | None = None):
def success(
*, code: int = 200, msg: str = 'Success', data: Any | None = None, exclude: _ExcludeData | None = None, **kwargs
) -> dict:
"""
请求成功返回通用方法
@ -41,14 +73,16 @@ class ResponseBase:
:param exclude: 排除返回数据(data)字段
:return:
"""
data = data if data is None else ResponseBase.__encode_json(data)
return ResponseModel(code=code, msg=msg, data=data).dict(exclude={'data': exclude})
data = data if data is None else ResponseBase.__json_encoder(data, exclude, **kwargs)
return {'code': code, 'msg': msg, 'data': data}
@staticmethod
@validate_arguments
def fail(*, code: int = 400, msg: str = 'Bad Request', data: Any = None, exclude: _JsonEncoder | None = None):
data = data if data is None else ResponseBase.__encode_json(data)
return ResponseModel(code=code, msg=msg, data=data).dict(exclude={'data': exclude})
def fail(
*, code: int = 400, msg: str = 'Bad Request', data: Any = None, exclude: _ExcludeData | None = None, **kwargs
) -> dict:
data = data if data is None else ResponseBase.__json_encoder(data, exclude, **kwargs)
return {'code': code, 'msg': msg, 'data': data}
response_base = ResponseBase()