Skip to content

Parameter checking plugin

Pait is based on Pydantic to perform parameter checking and type conversion for each parameter, which cannot satisfy the need for multiple parameter dependency checking. For this reason, Pait provides two kinds of parameter dependency checking functions through post plugins Required and AtMostOneOf.

1.Required Plugin

In the creation of route functions, often encounter some parameter dependencies, such as having a request parameter A, B, C, which B and C are optional and the requirement that B exists, C also needs to exist, B does not exist, C can not exist. At this point, you can use the Required plugin for parameter restriction, the following code:

docs_source_code/plugin/param_plugin/flask_with_required_plugin_demo.py
 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
from typing import Optional

from flask import Flask, Response, jsonify

from pait import field
from pait.app.flask import pait
from pait.exceptions import TipException
from pait.plugin.required import RequiredPlugin


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


@pait(post_plugin_list=[RequiredPlugin.build(required_dict={"email": ["user_name"]})])
def demo(
    uid: str = field.Query.i(),
    user_name: Optional[str] = field.Query.i(default=None),
    email: Optional[str] = field.Query.i(default=None),
) -> Response:
    return jsonify({"uid": uid, "user_name": user_name, "email": email})


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/param_plugin/starlette_with_required_plugin_demo.py
 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
from typing import Optional

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

from pait import field
from pait.app.starlette import pait
from pait.exceptions import TipException
from pait.plugin.required import RequiredPlugin


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


@pait(post_plugin_list=[RequiredPlugin.build(required_dict={"email": ["user_name"]})])
async def demo(
    uid: str = field.Query.i(),
    user_name: Optional[str] = field.Query.i(default=None),
    email: Optional[str] = field.Query.i(default=None),
) -> JSONResponse:
    return JSONResponse({"uid": uid, "user_name": user_name, "email": email})


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/param_plugin/sanic_with_required_plugin_demo.py
 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
from typing import Optional

from sanic import Request, Sanic
from sanic.response import HTTPResponse, json

from pait import field
from pait.app.sanic import pait
from pait.exceptions import TipException
from pait.plugin.required import RequiredPlugin


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


@pait(post_plugin_list=[RequiredPlugin.build(required_dict={"email": ["user_name"]})])
async def demo(
    uid: str = field.Query.i(),
    user_name: Optional[str] = field.Query.i(default=None),
    email: Optional[str] = field.Query.i(default=None),
) -> HTTPResponse:
    return json({"uid": uid, "user_name": user_name, "email": email})


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/param_plugin/tornado_with_required_plugin_demo.py
 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
from typing import Optional

from tornado.ioloop import IOLoop
from tornado.web import Application, RequestHandler

from pait import field
from pait.app.tornado import pait
from pait.exceptions import TipException
from pait.plugin.required import RequiredPlugin


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(post_plugin_list=[RequiredPlugin.build(required_dict={"email": ["user_name"]})])
    async def get(
        self,
        uid: str = field.Query.i(),
        user_name: Optional[str] = field.Query.i(default=None),
        email: Optional[str] = field.Query.i(default=None),
    ) -> None:
        self.write({"uid": uid, "user_name": user_name, "email": email})


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


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

The parameter uid is a required parameter in the route function, while the parameters user_name and email are optional, but a new validation rule is added when the ReuiredPlugin plugin is used. This validation rule is defined by required_dict, which states that the parameter email must depend on a collection of parameters to exist, and that collection has only one parameter -- user_name. So the validation rule for RequiredPlugin is that the parameter email can only exist if the parameter user_name exists.

After using curl to send a request through the response results can be found, if the request parameter is only uid can be returned normally, but the request parameter user_name is null, the parameter email must be null, otherwise it will report an error. 

➜ ~ curl http://127.0.0.1:8000/api/demo\?uid\=123
{"uid":"123","user_name":null,"email":null}%
➜ ~ curl http://127.0.0.1:8000/api/demo\?uid\=123\&email\=aaa
{"data":"email requires param user_name, which if not none"}%
➜ ~ curl http://127.0.0.1:8000/api/demo\?uid\=123\&email\=aaa\&user_name\=so1n
{"uid":"123","user_name":"so1n","email":"aaa"}%

The Required plugin can pass dependency rules through the build method, but it can also define rules through the ExtraParam extension parameter. The Required plugin supports both RequiredExtraParam and RequiredGroupExtraParam extension parameter. The following code is a use of RequiredExtraParam, which generates a validation rule for user_name dependent on email via extra_param_list=[RequiredExtraParam(main_column="email").

docs_source_code/plugin/param_plugin/flask_with_required_plugin_and_extra_param_demo.py
 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
from typing import Optional

from flask import Flask, Response, jsonify

from pait import field
from pait.app.flask import pait
from pait.exceptions import TipException
from pait.plugin.required import RequiredExtraParam, RequiredPlugin


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


@pait(post_plugin_list=[RequiredPlugin.build()])
def demo(
    uid: str = field.Query.i(),
    user_name: Optional[str] = field.Query.i(default=None, extra_param_list=[RequiredExtraParam(main_column="email")]),
    email: Optional[str] = field.Query.i(default=None),
) -> Response:
    return jsonify({"uid": uid, "user_name": user_name, "email": email})


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/param_plugin/starlette_with_required_plugin_and_extra_param_demo.py
 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
from typing import Optional

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

from pait import field
from pait.app.starlette import pait
from pait.exceptions import TipException
from pait.plugin.required import RequiredExtraParam, RequiredPlugin


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


@pait(post_plugin_list=[RequiredPlugin.build()])
async def demo(
    uid: str = field.Query.i(),
    user_name: Optional[str] = field.Query.i(default=None, extra_param_list=[RequiredExtraParam(main_column="email")]),
    email: Optional[str] = field.Query.i(default=None),
) -> JSONResponse:
    return JSONResponse({"uid": uid, "user_name": user_name, "email": email})


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/param_plugin/sanic_with_required_plugin_and_extra_param_demo.py
 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
from typing import Optional

from sanic import Request, Sanic
from sanic.response import HTTPResponse, json

from pait import field
from pait.app.sanic import pait
from pait.exceptions import TipException
from pait.plugin.required import RequiredExtraParam, RequiredPlugin


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


@pait(post_plugin_list=[RequiredPlugin.build()])
async def demo(
    uid: str = field.Query.i(),
    user_name: Optional[str] = field.Query.i(default=None, extra_param_list=[RequiredExtraParam(main_column="email")]),
    email: Optional[str] = field.Query.i(default=None),
) -> HTTPResponse:
    return json({"uid": uid, "user_name": user_name, "email": email})


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/param_plugin/tornado_with_required_plugin_and_extra_param_demo.py
 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
from typing import Optional

from tornado.ioloop import IOLoop
from tornado.web import Application, RequestHandler

from pait import field
from pait.app.tornado import pait
from pait.exceptions import TipException
from pait.plugin.required import RequiredExtraParam, RequiredPlugin


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(post_plugin_list=[RequiredPlugin.build()])
    async def get(
        self,
        uid: str = field.Query.i(),
        email: Optional[str] = field.Query.i(default=None),
        user_name: Optional[str] = field.Query.i(
            default=None, extra_param_list=[RequiredExtraParam(main_column="email")]
        ),
    ) -> None:
        self.write({"uid": uid, "user_name": user_name, "email": email})


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


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

Another extension parameter RequiredGroupExtraParam is to categorize the parameters by group and mark one of the parameters in this group as the main parameter by is_main, so that all other parameters in the group will depend on the main parameter. The following sample code categorizes user_name and email parameters into my-group, and defines the email parameter as the main parameter of my-group, so that the generated validation rules depend on the user_name parameter and the email parameter.

docs_source_code/plugin/param_plugin/flask_with_required_plugin_and_group_extra_param_demo.py
 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
from typing import Optional

from flask import Flask, Response, jsonify

from pait import field
from pait.app.flask import pait
from pait.exceptions import TipException
from pait.plugin.required import RequiredGroupExtraParam, RequiredPlugin


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


@pait(post_plugin_list=[RequiredPlugin.build()])
def demo(
    uid: str = field.Query.i(),
    user_name: Optional[str] = field.Query.i(
        default=None, extra_param_list=[RequiredGroupExtraParam(group="my-group")]
    ),
    email: Optional[str] = field.Query.i(
        default=None, extra_param_list=[RequiredGroupExtraParam(group="my-group", is_main=True)]
    ),
) -> Response:
    return jsonify({"uid": uid, "user_name": user_name, "email": email})


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/param_plugin/starlette_with_required_plugin_and_group_extra_param_demo.py
 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
from typing import Optional

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

from pait import field
from pait.app.starlette import pait
from pait.exceptions import TipException
from pait.plugin.required import RequiredGroupExtraParam, RequiredPlugin


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


@pait(post_plugin_list=[RequiredPlugin.build()])
async def demo(
    uid: str = field.Query.i(),
    user_name: Optional[str] = field.Query.i(
        default=None, extra_param_list=[RequiredGroupExtraParam(group="my-group")]
    ),
    email: Optional[str] = field.Query.i(
        default=None, extra_param_list=[RequiredGroupExtraParam(group="my-group", is_main=True)]
    ),
) -> JSONResponse:
    return JSONResponse({"uid": uid, "user_name": user_name, "email": email})


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/param_plugin/sanic_with_required_plugin_and_group_extra_param_demo.py
 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
from typing import Optional

from sanic import Request, Sanic
from sanic.response import HTTPResponse, json

from pait import field
from pait.app.sanic import pait
from pait.exceptions import TipException
from pait.plugin.required import RequiredGroupExtraParam, RequiredPlugin


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


@pait(post_plugin_list=[RequiredPlugin.build()])
async def demo(
    uid: str = field.Query.i(),
    user_name: Optional[str] = field.Query.i(
        default=None, extra_param_list=[RequiredGroupExtraParam(group="my-group")]
    ),
    email: Optional[str] = field.Query.i(
        default=None, extra_param_list=[RequiredGroupExtraParam(group="my-group", is_main=True)]
    ),
) -> HTTPResponse:
    return json({"uid": uid, "user_name": user_name, "email": email})


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/param_plugin/tornado_with_required_plugin_and_group_extra_param_demo.py
 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
from typing import Optional

from tornado.ioloop import IOLoop
from tornado.web import Application, RequestHandler

from pait import field
from pait.app.tornado import pait
from pait.exceptions import TipException
from pait.plugin.required import RequiredGroupExtraParam, RequiredPlugin


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(post_plugin_list=[RequiredPlugin.build()])
    async def get(
        self,
        uid: str = field.Query.i(),
        user_name: Optional[str] = field.Query.i(
            default=None, extra_param_list=[RequiredGroupExtraParam(group="my-group")]
        ),
        email: Optional[str] = field.Query.i(
            default=None, extra_param_list=[RequiredGroupExtraParam(group="my-group", is_main=True)]
        ),
    ) -> None:
        self.write({"uid": uid, "user_name": user_name, "email": email})


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


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

2.AtMostOneOf Plugin

The main function of the AtMostOneOf plugin is to verify whether the parameters are mutually exclusive. for example, if there are three parameters A, B and C and the B parameter is required to be mutually exclusive with the C parameter, that if B exists, C cannot exist, and if C exists, B cannot exist. This can use AtMostOneOf plugin configuration rules to achieve the function, the code is as follows:

docs_source_code/plugin/param_plugin/flask_with_at_most_one_of_plugin_demo.py
 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
from typing import Optional

from flask import Flask, Response, jsonify

from pait import field
from pait.app.flask import pait
from pait.exceptions import TipException
from pait.plugin.at_most_one_of import AtMostOneOfPlugin


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


@pait(post_plugin_list=[AtMostOneOfPlugin.build(at_most_one_of_list=[["email", "user_name"]])])
def demo(
    uid: str = field.Query.i(),
    user_name: Optional[str] = field.Query.i(default=None),
    email: Optional[str] = field.Query.i(default=None),
) -> Response:
    return jsonify({"uid": uid, "user_name": user_name, "email": email})


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/param_plugin/starlette_with_at_most_one_of_plugin_demo.py
 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
from typing import Optional

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

from pait import field
from pait.app.starlette import pait
from pait.exceptions import TipException
from pait.plugin.at_most_one_of import AtMostOneOfPlugin


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


@pait(post_plugin_list=[AtMostOneOfPlugin.build(at_most_one_of_list=[["email", "user_name"]])])
async def demo(
    uid: str = field.Query.i(),
    user_name: Optional[str] = field.Query.i(default=None),
    email: Optional[str] = field.Query.i(default=None),
) -> JSONResponse:
    return JSONResponse({"uid": uid, "user_name": user_name, "email": email})


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/param_plugin/sanic_with_at_most_one_of_plugin_demo.py
 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
from typing import Optional

from sanic import Request, Sanic
from sanic.response import HTTPResponse, json

from pait import field
from pait.app.sanic import pait
from pait.exceptions import TipException
from pait.plugin.at_most_one_of import AtMostOneOfPlugin


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


@pait(post_plugin_list=[AtMostOneOfPlugin.build(at_most_one_of_list=[["email", "user_name"]])])
async def demo(
    uid: str = field.Query.i(),
    user_name: Optional[str] = field.Query.i(default=None),
    email: Optional[str] = field.Query.i(default=None),
) -> HTTPResponse:
    return json({"uid": uid, "user_name": user_name, "email": email})


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/param_plugin/tornado_with_at_most_one_of_plugin_demo.py
 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
from typing import Optional

from tornado.ioloop import IOLoop
from tornado.web import Application, RequestHandler

from pait import field
from pait.app.tornado import pait
from pait.exceptions import TipException
from pait.plugin.at_most_one_of import AtMostOneOfPlugin


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(post_plugin_list=[AtMostOneOfPlugin.build(at_most_one_of_list=[["email", "user_name"]])])
    async def get(
        self,
        uid: str = field.Query.i(),
        user_name: Optional[str] = field.Query.i(default=None),
        email: Optional[str] = field.Query.i(default=None),
    ) -> None:
        self.write({"uid": uid, "user_name": user_name, "email": email})


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


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

In the sample code, uid is a required parameter, while user_name and email are optional parameters, and after using the AtMostOneOfPlugin plugin a new validation rule will be added. This validation rule is defined by the parameter at_most_one_of_list, which indicates that the parameters email and user_name cannot exist at the same time.

After sending a request using curl, the response shows that an error is returned when both email and user_name are present, but otherwise the response is returned normally. 

➜ ~ curl http://127.0.0.1:8000/api/demo\?uid\=123
{"uid":"123","user_name":null,"email":null}%
➜ ~ curl http://127.0.0.1:8000/api/demo\?uid\=123\&email\=aaa
{"uid":"123","user_name":null,"email":"aaa"}%
➜  ~ curl http://127.0.0.1:8000/api/demo\?uid\=123\&user_name\=so1n
{"uid":"123","user_name":"so1n","email":null}%
➜ ~ curl http://127.0.0.1:8000/api/demo\?uid\=123\&email\=aaa\&user_name\=so1n
{"data":"requires at most one of param email or user_name"}%

In addition, the AtMostOneOf plugin also supports grouping parameters by ExtraParam and restricting them to not appearing at the same time, using the following method:

docs_source_code/plugin/param_plugin/flask_with_at_most_one_of_plugin_and_extra_param_demo.py
 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
from typing import Optional

from flask import Flask, Response, jsonify

from pait import field
from pait.app.flask import pait
from pait.exceptions import TipException
from pait.plugin.at_most_one_of import AtMostOneOfExtraParam, AtMostOneOfPlugin


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


@pait(post_plugin_list=[AtMostOneOfPlugin.build()])
def demo(
    uid: str = field.Query.i(),
    email: Optional[str] = field.Query.i(default=None, extra_param_list=[AtMostOneOfExtraParam(group="my-group")]),
    user_name: Optional[str] = field.Query.i(default=None, extra_param_list=[AtMostOneOfExtraParam(group="my-group")]),
) -> Response:
    return jsonify({"uid": uid, "user_name": user_name, "email": email})


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/param_plugin/starlette_with_at_most_one_of_plugin_and_extra_param_demo.py
 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
from typing import Optional

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

from pait import field
from pait.app.starlette import pait
from pait.exceptions import TipException
from pait.plugin.at_most_one_of import AtMostOneOfExtraParam, AtMostOneOfPlugin


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


@pait(post_plugin_list=[AtMostOneOfPlugin.build()])
async def demo(
    uid: str = field.Query.i(),
    email: Optional[str] = field.Query.i(default=None, extra_param_list=[AtMostOneOfExtraParam(group="my-group")]),
    user_name: Optional[str] = field.Query.i(default=None, extra_param_list=[AtMostOneOfExtraParam(group="my-group")]),
) -> JSONResponse:
    return JSONResponse({"uid": uid, "user_name": user_name, "email": email})


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/param_plugin/sanic_with_at_most_one_of_plugin_and_extra_param_demo.py
 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
from typing import Optional

from sanic import Request, Sanic
from sanic.response import HTTPResponse, json

from pait import field
from pait.app.sanic import pait
from pait.exceptions import TipException
from pait.plugin.at_most_one_of import AtMostOneOfExtraParam, AtMostOneOfPlugin


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


@pait(post_plugin_list=[AtMostOneOfPlugin.build()])
async def demo(
    uid: str = field.Query.i(),
    email: Optional[str] = field.Query.i(default=None, extra_param_list=[AtMostOneOfExtraParam(group="my-group")]),
    user_name: Optional[str] = field.Query.i(default=None, extra_param_list=[AtMostOneOfExtraParam(group="my-group")]),
) -> HTTPResponse:
    return json({"uid": uid, "user_name": user_name, "email": email})


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/param_plugin/tornado_with_at_most_one_of_plugin_and_extra_param_demo.py
 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
from typing import Optional

from tornado.ioloop import IOLoop
from tornado.web import Application, RequestHandler

from pait import field
from pait.app.tornado import pait
from pait.exceptions import TipException
from pait.plugin.at_most_one_of import AtMostOneOfExtraParam, AtMostOneOfPlugin


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


class DemoHandler(_Handler):
    @pait(post_plugin_list=[AtMostOneOfPlugin.build()])
    async def get(
        self,
        uid: str = field.Query.i(),
        user_name: Optional[str] = field.Query.i(
            default=None, extra_param_list=[AtMostOneOfExtraParam(group="my-group")]
        ),
        email: Optional[str] = field.Query.i(default=None, extra_param_list=[AtMostOneOfExtraParam(group="my-group")]),
    ) -> None:
        self.write({"uid": uid, "user_name": user_name, "email": email})


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


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

In this code, the user_name and email parameters are grouped into my-group using AtMostOneOfExtraParam. At runtime, the AtMostOneOf plugin verifies that both the user_name and email parameters exist, and throws an error if both exist.