跳转至

参数检查插件

Pait基于Pydantic为每个参数进行参数校验和类型转换，无法满足多个参数依赖校验的需求， 为此，Pait通过后置插件Required和AtMostOneOf提供两种参数依赖校验功能。

1.Required插件

在创建路由函数时，经常会遇到一些参数依赖情况，比如拥有请求参数A，B，C， 其中，B和C都是选填，且要求B存在时，C也需要存在，B不存在时，C就不能存在。 这时可以使用Required插件来进行参数限制，如下代码：

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

路由函数中参数uid为必填参数，而参数user_name和email是选填参数，但是在使用ReuiredPlugin插件后会新增一个验证规则。 这个验证规则是由required_dict定义的，它表示参数email必须依赖于一个参数集合才可以存在，该集合只有一个参数--user_name， 所以RequiredPlugin的验证规则是是参数user_name存在的时候，参数email才可以存在。

使用curl发送请求后可以通过响应结果发现，如果请求的参数只有uid时能正常返回，但请求的参数user_name为空时，参数email必须为空，不然会报错。

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

Required插件除了通过build方法传递依赖规则外，也可以通过ExtraParam拓展参数来定义规则，Required插件支持RequiredExtraParam和RequiredGroupExtraParam两种拓展参数。 如下代码是RequiredExtraParam的使用通过向user_name参数的Field添加extra_param_list=[RequiredExtraParam(main_column="email")配置标记user_name依赖于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()

另一个拓展参数RequiredGroupExtraParam则是通过group为参数做分类并通过is_main标记这组参数中的某个参数为主参数，这样一来该分组的其他参数都会依赖到主参数。 如下示例代码把user_name和email参数归为my-group组，同时定义email参数为my-group组的主参数，最终生成的验证规则依赖就是user_name参数依赖于email参数。

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插件

AtMostOneOf插件的主要功能是验证参数是否互斥，比如有三个参数A、B和C且要求B参数与C参数互斥，也就是B存在时，C就不能存在，C存在时，B就不能存在。 这时可以使用AtMostOneOf插件配置规则来实现功能，代码如下：

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

在示例代码中， uid为必填参数，而user_name和email是选填参数，在使用AtMostOneOfPlugin插件后就会新增一条验证规则， 这条验证规则是由参数at_most_one_of_list定义的，它表示的是参数email和user_name不能同时存在。

在使用curl发送请求后，通过响应结果可以发现参数email和user_name共存时候会返回错误，其它情况都能正常返回响应。

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

此外，AtMostOneOf插件也支持通过ExtraParam对参数进行归类，并限制它们不能同时出现，使用方法如下：

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

在这段代码中，使用AtMostOneOfExtraParam把user_name和email参数归为my-group组。 在运行时，AtMostOneOf插件会验证user_name和email参数是否都存在，如果同时存在则会直接抛出错误。