Skip to content

Json Response plugin

目前API接口用的最多的序列化方式就是Json，所以基于Json响应有很多特别的需求，目前，Pait自带了几个与Json响应相关的插件，他们都用到了Pait装饰器填写的response_model_list。

Note

  • 1.由于要获取到返回的结果，所以这两个插件都会侵入到原有框架，导致使用方法与原有框架有些不同。
  • 2.这几个插件都要单独根据不同的Web框架进行兼容，所以请以from pait.app.{web framework name}.plugin.{plugin name} import xxx的形式来引入对应的插件。

校验Json响应结果插件

校验Json响应结果插件的主要功能是在收到返回响应结果时，对响应结果进行校验，如果校验成功，才会返回响应，否则就会报错。 以example.param_verify.starlette_example.async_check_json_plugin_route为例子： 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
from typing import Optional
from typing_extensions import TypedDict  # 对于Python3.8以下的只能通过typing_extensions引入
import uvicorn  # type: ignore
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route
from pait.exceptions import TipException
from pait.app.starlette.plugin.check_json_resp import CheckJsonRespPlugin
from example.param_verify.model import UserSuccessRespModel3

from pait.app.starlette import pait
from pait import field


async def api_exception(request: Request, exc: Exception) -> JSONResponse:
    """插件校验不通过会直接抛出异常，该函数会提取异常信息，并以返回对应的错误信息"""
    if isinstance(exc, TipException):
        exc = exc.exc
    return JSONResponse({"data": str(exc)})


_sub_typed_dict = TypedDict(
    "_sub_typed_dict",
    {
        "uid": int,
        "user_name": str,
        "email": str,
    },
)
_typed_dict = TypedDict(
    "_typed_dict",
    {
        "code": int,
        "msg": str,
        "data": _sub_typed_dict,
    },
)


@pait(response_model_list=[UserSuccessRespModel3], plugin_list=[CheckJsonRespPlugin.build()])
async def demo(
    uid: int = field.Query.i(description="user id", gt=10, lt=1000),
    email: Optional[str] = field.Query.i(default="example@xxx.com", description="user email"),
    user_name: str = field.Query.i(description="user name", min_length=2, max_length=4),
    age: int = field.Query.i(description="age", gt=1, lt=100),
    display_age: int = field.Query.i(0, description="display_age"),
) -> _typed_dict:
    """Test json plugin by resp type is typed dict"""
    return_dict: dict = {
        "code": 0,
        "msg": "",
        "data": {
            "uid": uid,
            "user_name": user_name,
            "email": email,
        },
    }
    if display_age == 1:
        return_dict["data"]["age"] = age
    return return_dict  # type: ignore


app = Starlette(routes=[Route("/api/demo", demo, methods=["GET"])])
app.add_exception_handler(Exception, api_exception)
uvicorn.run(app)
在这个代码中，首先是在23-38行定义了一个名为_typed_dict类型，他的结构与UserSuccessRespModel3.response_data一样， 这个数据结构会在demo函数中使用，定义demo函数的返回类型为 _typed_dict。

如果觉得重复定义会比较麻烦，可以直接填写为dict， 但是这样在编写代码时类型检查工具就没办法检查返回的数据结构是否正确了。

然后在41行中引入了一个名为CheckJsonRespPlugin的插件，该插件会在启动的时候检查路由函数定义的返回类型与UserSuccessRespModel3.response_data是否一致，不一致则会报错。 然后它在运行时校验路由函数响应的字典结构中每个字段的类型是否与UserSuccessRespModel3.response_data一致，如果校验失败则返回错误，校验成功则调用Web框架对应的Json响应对象把数据返回给客户端。 具体示例如下： 

➜  ~ curl http://127.0.0.1:8000/api/demo\?uid\=123\&user_name\=so1n\&age\=18
{"data":"1 validation error for ResponseModel\ndata -> age\n  field required (type=value_error.missing)"}%
➜  ~ curl http://127.0.0.1:8000/api/demo\?uid\=123\&user_name\=so1n\&age\=18\&display_age\=1
{"code":0,"msg":"","data":{"uid":123,"user_name":"so1n","email":"example@xxx.com","age":18}}%
通过结果可以发现，当响应结果与定义的响应Model不匹配时，会直接抛出错误，匹配则正常响应。

自动补全Json响应结果插件

在编写API接口的时候，接口返回的响应结果应该会与文档描述的保持一致，但可能会因为一些筛选条件的不同经常导致返回的响应结果是定义响应模型的子集，这种情况下如果客户端没有做特殊处理就会抛出异常，这时可以采用自动补全Json响应结果插件，自动为那些缺少的字段补上默认值。

以上面的代码为例子，去掉变量_typed_dict，再把插件CheckJsonRespPlugin替换为AutoCompleteJsonRespPlugin，代码如下: 

from typing import Optional
import uvicorn  # type: ignore
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route
from pait.exceptions import TipException
from pait.app.starlette.plugin.auto_complete_json_resp import AutoCompleteJsonRespPlugin
from example.param_verify.model import UserSuccessRespModel3

from pait.app.starlette import pait
from pait import field


async def api_exception(request: Request, exc: Exception) -> JSONResponse:
    """提取异常信息， 并以响应返回"""
    if isinstance(exc, TipException):
        exc = exc.exc
    return JSONResponse({"data": str(exc)})


@pait(
    response_model_list=[UserSuccessRespModel3],
    plugin_list=[AutoCompleteJsonRespPlugin.build()]
)
async def demo(
    uid: int = field.Query.i(description="user id", gt=10, lt=1000),
    email: Optional[str] = field.Query.i(default="example@xxx.com", description="user email"),
    user_name: str = field.Query.i(description="user name", min_length=2, max_length=4),
    age: int = field.Query.i(description="age", gt=1, lt=100),
    display_age: int = field.Query.i(0, description="display_age"),
) -> dict:
    """Test json plugin by resp type is typed dict"""
    return_dict: dict = {
        "code": 0,
        "msg": "",
        "data": {
            "uid": uid,
            "user_name": user_name,
            "email": email,
        },
    }
    if display_age == 1:
        return_dict["data"]["age"] = age
    return return_dict  # type: ignore


app = Starlette(routes=[Route("/api/demo", demo, methods=["GET"])])
app.add_exception_handler(Exception, api_exception)
uvicorn.run(app)
接着再运行与上面例子相同的请求: 
➜  ~ curl http://127.0.0.1:8000/api/demo\?uid\=123\&user_name\=so1n\&age\=18\&display_age\=1
{"code":0,"msg":"","data":{"uid":123,"user_name":"so1n","age":18,"email":"example@xxx.com"}}%
➜  ~ curl http://127.0.0.1:8000/api/demo\?uid\=123\&user_name\=so1n\&age\=18
{"code":0,"msg":"","data":{"uid":123,"user_name":"so1n","age":0,"email":"example@xxx.com"}}%
通过响应结果可以发现，对于第一个响应结果的age值为调用命令时填写的18，而第二个响应结果中本来是没有age字段的，该字段值是由插件AutoCompleteJsonRespPlugin根据age的类型自动填上默认值0。

AutoCompleteJsonRespPlugin自动补全的原理是从response_model_list中选出开发者填写的第一个ResponseModel，比如代码中的例子就是UserSuccessRespModel3， 然后通过调用UserSuccessRespModel3的get_default_dict获取到对应的默认值，再与路由函数返回的数据结构进行对比，如果发现响应的数据结构缺少对应的字段，就会自动补上。 如果开发者觉得自动生成的默认值不喜欢，那么可以通过字段对应的Field来指定自己想要的默认值，比如对UserSuccessRespModel3进行更改: 

class UserSuccessRespModel3(PaitJsonResponseModel):
    is_core: bool = True

    class ResponseModel(ResponseModel):  # type: ignore
        class DataModel(BaseModel):
            uid: int = Field(description="user id", gt=10, lt=1000)
            user_name: str = Field(description="user name", min_length=2, max_length=4)
            age: int = Field(default=10, description="age", gt=1, lt=100)
            email: str = Field(description="user email")

        data: DataModel

    description: str = "success response"
    response_data: Type[BaseModel] = ResponseModel
通过Field定义age的默认值为10,再运行一样的请求后可以可以发现，返回的age默认值变为10： 
➜  ~ curl http://127.0.0.1:8000/api/demo\?uid\=123\&user_name\=so1n\&age\=18\&display_age\=1
{"code":0,"msg":"","data":{"uid":123,"user_name":"so1n","age":18,"email":"example@xxx.com"}}%
➜  ~ curl http://127.0.0.1:8000/api/demo\?uid\=123\&user_name\=so1n\&age\=18
{"code":0,"msg":"","data":{"uid":123,"user_name":"so1n","age":10,"email":"example@xxx.com"}}%

Back to top