Json响应插件

目前路由函数中用得最多的序列化方式是JSON，所以Pait也自带了一些与JSON响应相关的插件，如校验JSON响应结果，自动补充JSON响应结果数据等，它们都用到了response_model_list中的响应模型拓展对应的功能。

Note

1.由于插件需要获取到返回的结果，所以插件有可能侵入到原有框架，导致使用方法与原先的用法有些不同。
2.插件都需要根据不同的Web框架进行适配，请以from pait.app.{web framework name}.plugin.{plugin name} import xxx的形式来引入对应的插件。

校验JSON响应结果插件

校验JSON响应结果插件的主要功能是对路由函数的响应结果进行校验，如果校验成功，才会返回响应，否则就会抛出错误，如下例子:

FlaskStarletteSanicTornado

docs_source_code/docs_source_code/plugin/json_plugin/flask_with_check_json_plugin_demo.py
from typing import Type

from flask import Flask, Response, jsonify
from pydantic import BaseModel, Field

from pait.app.flask import pait
from pait.app.flask.plugin import CheckJsonRespPlugin
from pait.exceptions import TipException
from pait.field import Query
from pait.model.response import JsonResponseModel


class UserSuccessRespModel3(JsonResponseModel):
    class ResponseModel(BaseModel):  # 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(description="age", gt=1, lt=100)
            email: str = Field(description="user email")

        code: int = Field(0, description="api code")
        msg: str = Field("success", description="api status msg")
        data: DataModel

    description: str = "success response"
    response_data: Type[BaseModel] = ResponseModel


def api_exception(exc: Exception) -> Response:
    if isinstance(exc, TipException):
        exc = exc.exc
    return jsonify({"data": str(exc)})


@pait(response_model_list=[UserSuccessRespModel3], plugin_list=[CheckJsonRespPlugin.build()])
def demo(
    uid: int = Query.i(description="user id", gt=10, lt=1000),
    email: str = Query.i(default="example@xxx.com", description="user email"),
    user_name: str = Query.i(description="user name", min_length=2, max_length=4),
    age: int = Query.i(description="age", gt=1, lt=100),
    display_age: int = Query.i(0, description="display_age"),
) -> Response:
    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 jsonify(return_dict)


app = Flask("demo")
app.add_url_rule("/api/demo", view_func=demo, methods=["GET"])
app.errorhandler(Exception)(api_exception)


if __name__ == "__main__":
    app.run(port=8000)

docs_source_code/docs_source_code/plugin/json_plugin/starlette_with_check_json_plugin_demo.py
from typing import Type

from pydantic import BaseModel, Field
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse, Response
from starlette.routing import Route

from pait.app.starlette import pait
from pait.app.starlette.plugin import CheckJsonRespPlugin
from pait.exceptions import TipException
from pait.field import Query
from pait.model.response import JsonResponseModel


class UserSuccessRespModel3(JsonResponseModel):
    class ResponseModel(BaseModel):  # 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(description="age", gt=1, lt=100)
            email: str = Field(description="user email")

        code: int = Field(0, description="api code")
        msg: str = Field("success", description="api status msg")
        data: DataModel

    description: str = "success response"
    response_data: Type[BaseModel] = ResponseModel


async def api_exception(request: Request, exc: Exception) -> Response:
    if isinstance(exc, TipException):
        exc = exc.exc
    return JSONResponse({"data": str(exc)})


@pait(response_model_list=[UserSuccessRespModel3], plugin_list=[CheckJsonRespPlugin.build()])
async def demo(
    uid: int = Query.i(description="user id", gt=10, lt=1000),
    email: str = Query.i(default="example@xxx.com", description="user email"),
    user_name: str = Query.i(description="user name", min_length=2, max_length=4),
    age: int = Query.i(description="age", gt=1, lt=100),
    display_age: int = Query.i(0, description="display_age"),
) -> Response:
    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 JSONResponse(return_dict)


app = Starlette(routes=[Route("/api/demo", demo, methods=["GET"])])
app.add_exception_handler(Exception, api_exception)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

docs_source_code/docs_source_code/plugin/json_plugin/sanic_with_check_json_plugin_demo.py
from typing import Type

from pydantic import BaseModel, Field
from sanic import HTTPResponse, Request, Sanic, response

from pait.app.sanic import pait
from pait.app.sanic.plugin import CheckJsonRespPlugin
from pait.exceptions import TipException
from pait.field import Query
from pait.model.response import JsonResponseModel


class UserSuccessRespModel3(JsonResponseModel):
    class ResponseModel(BaseModel):  # 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(description="age", gt=1, lt=100)
            email: str = Field(description="user email")

        code: int = Field(0, description="api code")
        msg: str = Field("success", description="api status msg")
        data: DataModel

    description: str = "success response"
    response_data: Type[BaseModel] = ResponseModel


async def api_exception(request: Request, exc: Exception) -> response.HTTPResponse:
    if isinstance(exc, TipException):
        exc = exc.exc
    return response.json({"data": str(exc)})


@pait(response_model_list=[UserSuccessRespModel3], plugin_list=[CheckJsonRespPlugin.build()])
async def demo(
    uid: int = Query.i(description="user id", gt=10, lt=1000),
    email: str = Query.i(default="example@xxx.com", description="user email"),
    user_name: str = Query.i(description="user name", min_length=2, max_length=4),
    age: int = Query.i(description="age", gt=1, lt=100),
    display_age: int = Query.i(0, description="display_age"),
) -> HTTPResponse:
    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 response.json(return_dict)


app = Sanic("demo")
app.add_route(demo, "/api/demo", methods=["GET"])
app.exception(Exception)(api_exception)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

docs_source_code/docs_source_code/plugin/json_plugin/tornado_with_check_json_plugin_demo.py
from typing import Type

from pydantic import BaseModel, Field
from tornado.ioloop import IOLoop
from tornado.web import Application, RequestHandler

from pait.app.tornado import pait
from pait.app.tornado.plugin import CheckJsonRespPlugin
from pait.exceptions import TipException
from pait.field import Query
from pait.model.response import JsonResponseModel


class UserSuccessRespModel3(JsonResponseModel):
    class ResponseModel(BaseModel):  # 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(description="age", gt=1, lt=100)
            email: str = Field(description="user email")

        code: int = Field(0, description="api code")
        msg: str = Field("success", description="api status msg")
        data: DataModel

    description: str = "success response"
    response_data: Type[BaseModel] = ResponseModel


class _Handler(RequestHandler):
    def _handle_request_exception(self, exc: BaseException) -> None:
        if isinstance(exc, TipException):
            exc = exc.exc

        self.write({"data": str(exc)})
        self.finish()


class DemoHandler(_Handler):
    @pait(response_model_list=[UserSuccessRespModel3], plugin_list=[CheckJsonRespPlugin.build()])
    async def get(
        self,
        uid: int = Query.i(description="user id", gt=10, lt=1000),
        email: str = Query.i(default="example@xxx.com", description="user email"),
        user_name: str = Query.i(description="user name", min_length=2, max_length=4),
        age: int = Query.i(description="age", gt=1, lt=100),
        display_age: int = Query.i(0, description="display_age"),
    ) -> None:
        return_dict: dict = {
            "code": 0,
            "msg": "",
            "data": {
                "uid": uid,
                "user_name": user_name,
                "email": email,
            },
        }
        if display_age == 1:
            return_dict["data"]["age"] = age
        self.write(return_dict)


app: Application = Application([(r"/api/demo", DemoHandler)])


if __name__ == "__main__":
    app.listen(8000)
    IOLoop.instance().start()

首先是定义了一个名为UserSuccessRespModel3的JSON响应结果Model，然后是定义一个错误处理函数用于捕获插件校验结果失败后抛出的异常，接着是定义一个demo路由函数，路由函数使用了CheckJsonRespPlugin插件。另外，当display_age不等于1时，demo路由函数返回的结果会与UserSuccessRespModel3不匹配。

在运行代码并执行如下命令，通过执行结果可以发现，当响应结果与定义的响应Model不匹配时，会直接抛出错误：

➜  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}}
➜  curl http://127.0.0.1:8000/api/demo\?uid\=123\&user_name\=so1n\&age\=18
1 validation error for ResponseModel
data -> age
  field required (type=value_error.missing)

自动补全JSON响应结果插件

路由函数返回的结果应该与API文档定义的结构体一致，因为只返回部分字段就有可能导致客户端发生崩溃。如果由于某些原因只能返回结构体的部分字段，那么可以采用自动补全JSON响应结果插件为缺少值的字段补上字段对应的默认值，如下例子:

FlaskStarletteSanicTornado

docs_source_code/docs_source_code/plugin/json_plugin/flask_with_auto_complete_json_plugin_demo.py
from typing import List, Type

from flask import Flask
from pydantic import BaseModel, Field

from pait.app.flask import pait
from pait.app.flask.plugin import AutoCompleteJsonRespPlugin
from pait.model.response import JsonResponseModel


class AutoCompleteRespModel(JsonResponseModel):
    class ResponseModel(BaseModel):
        class DataModel(BaseModel):
            class MusicModel(BaseModel):
                name: str = Field("")
                url: str = Field()
                singer: str = Field("")

            uid: int = Field(100, description="user id", gt=10, lt=1000)
            music_list: List[MusicModel] = Field(description="music list")
            image_list: List[dict] = Field(description="music list")

        code: int = Field(0, description="api code")
        msg: str = Field("success", description="api status msg")
        data: DataModel

    description: str = "success response"
    response_data: Type[BaseModel] = ResponseModel


@pait(response_model_list=[AutoCompleteRespModel], plugin_list=[AutoCompleteJsonRespPlugin.build()])
def demo() -> dict:
    """Test json plugin by resp type is dict"""
    return {
        "code": 0,
        "msg": "",
        "data": {
            # "uid": 0,
            "image_list": [
                {"aaa": 10},
                {"aaa": "123"},
            ],
            "music_list": [
                {
                    "name": "music1",
                    "url": "http://music1.com",
                    "singer": "singer1",
                },
                {
                    # "name": "music1",
                    "url": "http://music1.com",
                    # "singer": "singer1",
                },
            ],
        },
    }


app = Flask("demo")
app.add_url_rule("/api/demo", view_func=demo, methods=["GET"])


if __name__ == "__main__":
    app.run(port=8000)

docs_source_code/docs_source_code/plugin/json_plugin/starlette_with_auto_complete_json_plugin_demo.py
from typing import List, Type

from pydantic import BaseModel, Field
from starlette.applications import Starlette
from starlette.routing import Route

from pait.app.starlette import pait
from pait.app.starlette.plugin import AutoCompleteJsonRespPlugin
from pait.model.response import JsonResponseModel


class AutoCompleteRespModel(JsonResponseModel):
    class ResponseModel(BaseModel):
        class DataModel(BaseModel):
            class MusicModel(BaseModel):
                name: str = Field("")
                url: str = Field()
                singer: str = Field("")

            uid: int = Field(100, description="user id", gt=10, lt=1000)
            music_list: List[MusicModel] = Field(description="music list")
            image_list: List[dict] = Field(description="music list")

        code: int = Field(0, description="api code")
        msg: str = Field("success", description="api status msg")
        data: DataModel

    description: str = "success response"
    response_data: Type[BaseModel] = ResponseModel


@pait(response_model_list=[AutoCompleteRespModel], plugin_list=[AutoCompleteJsonRespPlugin.build()])
async def demo() -> dict:
    """Test json plugin by resp type is dict"""
    return {
        "code": 0,
        "msg": "",
        "data": {
            # "uid": 0,
            "image_list": [
                {"aaa": 10},
                {"aaa": "123"},
            ],
            "music_list": [
                {
                    "name": "music1",
                    "url": "http://music1.com",
                    "singer": "singer1",
                },
                {
                    # "name": "music1",
                    "url": "http://music1.com",
                    # "singer": "singer1",
                },
            ],
        },
    }


app = Starlette(routes=[Route("/api/demo", demo, methods=["GET"])])


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

docs_source_code/docs_source_code/plugin/json_plugin/sanic_with_auto_complete_json_plugin_demo.py
from typing import List, Type

from pydantic import BaseModel, Field
from sanic import Request, Sanic

from pait.app.sanic import pait
from pait.app.sanic.plugin import AutoCompleteJsonRespPlugin
from pait.model.response import JsonResponseModel


class AutoCompleteRespModel(JsonResponseModel):
    class ResponseModel(BaseModel):
        class DataModel(BaseModel):
            class MusicModel(BaseModel):
                name: str = Field("")
                url: str = Field()
                singer: str = Field("")

            uid: int = Field(100, description="user id", gt=10, lt=1000)
            music_list: List[MusicModel] = Field(description="music list")
            image_list: List[dict] = Field(description="music list")

        code: int = Field(0, description="api code")
        msg: str = Field("success", description="api status msg")
        data: DataModel

    description: str = "success response"
    response_data: Type[BaseModel] = ResponseModel


@pait(response_model_list=[AutoCompleteRespModel], plugin_list=[AutoCompleteJsonRespPlugin.build()])
async def demo(request: Request) -> dict:
    """Test json plugin by resp type is dict"""
    return {
        "code": 0,
        "msg": "",
        "data": {
            # "uid": 0,
            "image_list": [
                {"aaa": 10},
                {"aaa": "123"},
            ],
            "music_list": [
                {
                    "name": "music1",
                    "url": "http://music1.com",
                    "singer": "singer1",
                },
                {
                    # "name": "music1",
                    "url": "http://music1.com",
                    # "singer": "singer1",
                },
            ],
        },
    }


app = Sanic("demo")
app.add_route(demo, "/api/demo", methods=["GET"])


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

docs_source_code/docs_source_code/plugin/json_plugin/tornado_with_auto_complete_json_plugin_demo.py
from typing import List, Type

from pydantic import BaseModel, Field
from tornado.ioloop import IOLoop
from tornado.web import Application, RequestHandler

from pait.app.tornado import pait
from pait.app.tornado.plugin import AutoCompleteJsonRespPlugin
from pait.model.response import JsonResponseModel


class AutoCompleteRespModel(JsonResponseModel):
    class ResponseModel(BaseModel):
        class DataModel(BaseModel):
            class MusicModel(BaseModel):
                name: str = Field("")
                url: str = Field()
                singer: str = Field("")

            uid: int = Field(100, description="user id", gt=10, lt=1000)
            music_list: List[MusicModel] = Field(description="music list")
            image_list: List[dict] = Field(description="music list")

        code: int = Field(0, description="api code")
        msg: str = Field("success", description="api status msg")
        data: DataModel

    description: str = "success response"
    response_data: Type[BaseModel] = ResponseModel


class DemoHandler(RequestHandler):
    @pait(response_model_list=[AutoCompleteRespModel], plugin_list=[AutoCompleteJsonRespPlugin.build()])
    async def get(
        self,
    ) -> dict:
        """Test json plugin by resp type is dict"""
        return {
            "code": 0,
            "msg": "",
            "data": {
                # "uid": 0,
                "image_list": [
                    {"aaa": 10},
                    {"aaa": "123"},
                ],
                "music_list": [
                    {
                        "name": "music1",
                        "url": "http://music1.com",
                        "singer": "singer1",
                    },
                    {
                        # "name": "music1",
                        "url": "http://music1.com",
                        # "singer": "singer1",
                    },
                ],
            },
        }


app: Application = Application([(r"/api/demo", DemoHandler)])


if __name__ == "__main__":
    app.listen(8000)
    IOLoop.instance().start()

这段代码先定义了一个AutoCompleteRespModel响应Model，其中UID的默认值为100。接着再实现一个demo函数，这个demo函数的返回结构有部分字段是缺失的但是它使用了AutoCompleteJsonRespPlugin插件，接着运行代码并执行如下命令：

➜  ~ curl http://127.0.0.1:8000/api/demo
{
  "code":0,
  "data":{
      "image_list":[{},{}],
      "music_list":[{"name":"music1","singer":"singer1","url":"http://music1.com"},{"name":"","singer":"","url":"http://music1.com"}],
      "uid":100
    },
  "msg":""
}

通过输出的结果可以发现，data->uid，data->music_list->[0]->name以及data->music_list->[0]->singer都被补上了默认值，其中data->uid的默认值为AutoCompleteRespModel的Field定义的，而其他字段的默认值为类型对应的零值。

Note

1.通过Field的default或者是default_factory可以定义响应的默认值。 2.AutoCompletePlugin会侵入路由函数，导致路由函数只能返回Python类型而不是响应对象。