Json response plugin

Currently the most used serialization method in the route function is JSON, so Pait also comes with some JSON response related plug-ins, such as checking the JSON response result, automatically replenishing the JSON response result data, etc., which all use the response model_list in the response model to expand the corresponding functions.

Note

1.Since the plugin needs to get the returned results, the plugin may intrude into the web framework, resulting in a somewhat different usage than the original usage.
2.Plugins need to be adapted to different web frameworks, so please introduce the corresponding plugin in the form of from pait.app.{web framework name}.plugin.{plugin name} import xxx.

Check the JSON response result plugin

The CheckJsonPlugin plugin is mainly used to check the response result of the route function, if the check is successful, the response will be returned, otherwise an error will be thrown, as shown below:

FlaskStarletteSanicTornado

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/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/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/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()

The first step is to define a JSON response result model called UserSuccessRespModel3. Then define an error handler function that catches exceptions thrown when the plugin fails to validate the results. Next is to define a demo route function that uses the CheckJsonRespPlugin plugin. Also, when display_age is not equal to 1, the result returned by the demo route function will not match the UserSuccessRespModel3.

After running the code and executing the following commands, can find through the execution results that when the response result does not match the defined response Model, an error will be thrown directly:

➜  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)

Autocomplete JSON response result plugin

The result returned by the route function should be consistent with the structure defined in the API documentation, because returning only some of the fields may cause the client to crash. If for some reason only part of the structure can be returned, then can use the AutoComplete JSON Response plugin to fill in the default values for the fields that are missing, as in the following example:

FlaskStarletteSanicTornado

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/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/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/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()

First define an AutoCompleteRespModel response Model with a default value of 100 for the UID in the Model. Then create a demo function, which has some fields missing from the return structure, but uses the AutoCompleteJsonRespPlugin plugin.

Run the code and execute the following command:

➜  ~ 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":""
}

The output shows that data->uid, data->music_list->[0]->name and data->music_list->[0]->singer have been supplemented with default values. The default value of data->uid is defined for the Field of AutoCompleteRespModel, while the default values of the other fields are zero values corresponding to the type.

Note

1.The default value of the response can be defined through default or default_factory of Field. 2.AutoCompletePlugin intrudes into the routing function, causing the routing function to return only Python types instead of response objects.