如何使用Field对象

Field对象在Pait中起到了至关重要的作用， Pait除了通过Field对象获取数据来源外， 还可以实现其它的功能， 不过本章中只着重说明参数校验。

1.Field的种类

除了介绍提到的Body外， 还有其他不同含义的Field对象， 它们的名称和作用如下:

• Body: 获取当前请求的json数据
• Cookie: 获取当前请求的cookie数据(注意， 目前Cookie数据会被转化为Python的dict对象， 这意味着Cookie的Key不能重复。建议当Field为Cookie时，参数的类型为str)
• File：获取当前请求的file对象，该对象与原Web框架的file对象一致
• Form：获取当前请求的form数据，如果有多个重复Key，只会返回第一个值
• Header: 获取当前请求的header数据
• Json: 获取当前请求的json数据(与Body一样)
• Path: 获取当前请求的path数据，如/api/{version}/test，则会获取到version的数据
• Query: 获取当前请求的Url参数对应的数据，如果有多个重复Key，只会返回第一个值
• MultiForm：获取当前请求的form数据， 返回Key对应的数据列表
• MultiQuery：获取当前请求的Url参数对应的数据， 返回Key对应的数据列表

Field使用方法很简单，只要在<name>:<type>=<default>的default使用Field即可，以这段代码为例子:

FlaskStarletteSanicTornado

from enum import Enum
from typing import List, Optional

from flask import Flask

from pait.app.flask import pait
from pait.field import Cookie, Form, MultiForm, MultiQuery, Path, Query


class SexEnum(str, Enum):
    man: str = "man"
    woman: str = "woman"


@pait()
def demo_route(
    a: str = Form.t(description="form data"),
    b: str = Form.t(description="form data"),
    c: List[str] = MultiForm.t(description="form data"),
    cookie: dict = Cookie.t(raw_return=True, description="cookie"),
    multi_user_name: List[str] = MultiQuery.t(description="user name", min_length=2, max_length=4),
    age: int = Path.t(description="age", gt=1, lt=100),
    uid: int = Query.t(description="user id", gt=10, lt=1000),
    user_name: str = Query.t(description="user name", min_length=2, max_length=4),
    email: Optional[str] = Query.t(default="example@xxx.com", description="user email"),
    sex: SexEnum = Query.t(description="sex"),
) -> dict:
    return {
        "code": 0,
        "msg": "",
        "data": {
            "form_a": a,
            "form_b": b,
            "form_c": c,
            "cookie": cookie,
            "multi_user_name": multi_user_name,
            "age": age,
            "uid": uid,
            "user_name": user_name,
            "email": email,
            "sex": sex,
        },
    }


app = Flask("demo")
app.add_url_rule("/api/demo/<age>", "demo", demo_route, methods=["POST"])


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

from enum import Enum
from typing import List, Optional

from starlette.applications import Starlette
from starlette.responses import JSONResponse

from pait.app.starlette import pait
from pait.field import Cookie, Form, MultiForm, MultiQuery, Path, Query


class SexEnum(str, Enum):
    man: str = "man"
    woman: str = "woman"


@pait()
async def demo_route(
    a: str = Form.t(description="form data"),
    b: str = Form.t(description="form data"),
    c: List[str] = MultiForm.t(description="form data"),
    cookie: dict = Cookie.t(raw_return=True, description="cookie"),
    multi_user_name: List[str] = MultiQuery.t(description="user name", min_length=2, max_length=4),
    age: int = Path.t(description="age", gt=1, lt=100),
    uid: int = Query.t(description="user id", gt=10, lt=1000),
    user_name: str = Query.t(description="user name", min_length=2, max_length=4),
    email: Optional[str] = Query.t(default="example@xxx.com", description="user email"),
    sex: SexEnum = Query.t(description="sex"),
) -> JSONResponse:
    return JSONResponse(
        {
            "code": 0,
            "msg": "",
            "data": {
                "form_a": a,
                "form_b": b,
                "form_c": c,
                "cookie": cookie,
                "multi_user_name": multi_user_name,
                "age": age,
                "uid": uid,
                "user_name": user_name,
                "email": email,
                "sex": sex,
            },
        }
    )


app: Starlette = Starlette()
app.add_route("/api/demo/{age}", demo_route, methods=["POST"])


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

from enum import Enum
from typing import List, Optional

from sanic import HTTPResponse, Sanic, json

from pait.app.sanic import pait
from pait.field import Cookie, Form, MultiForm, MultiQuery, Path, Query


class SexEnum(str, Enum):
    man: str = "man"
    woman: str = "woman"


@pait()
async def demo_route(
    a: str = Form.t(description="form data"),
    b: str = Form.t(description="form data"),
    c: List[str] = MultiForm.t(description="form data"),
    cookie: dict = Cookie.t(raw_return=True, description="cookie"),
    multi_user_name: List[str] = MultiQuery.t(description="user name", min_length=2, max_length=4),
    age: int = Path.t(description="age", gt=1, lt=100),
    uid: int = Query.t(description="user id", gt=10, lt=1000),
    user_name: str = Query.t(description="user name", min_length=2, max_length=4),
    email: Optional[str] = Query.t(default="example@xxx.com", description="user email"),
    sex: SexEnum = Query.t(description="sex"),
) -> HTTPResponse:
    return json(
        {
            "code": 0,
            "msg": "",
            "data": {
                "form_a": a,
                "form_b": b,
                "form_c": c,
                "cookie": cookie,
                "multi_user_name": multi_user_name,
                "age": age,
                "uid": uid,
                "user_name": user_name,
                "email": email,
                "sex": sex,
            },
        }
    )


app: Sanic = Sanic(name="demo")
app.add_route(demo_route, "/api/demo/<age>", methods={"POST"})


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

from enum import Enum
from typing import List, Optional

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

from pait.app.tornado import pait
from pait.field import Cookie, Form, MultiForm, MultiQuery, Path, Query
from pait.openapi.doc_route import AddDocRoute


class SexEnum(str, Enum):
    man: str = "man"
    woman: str = "woman"


class DemoHandler(RequestHandler):
    @pait()
    async def post(
        self,
        a: str = Form.t(description="form data"),
        b: str = Form.t(description="form data"),
        c: List[str] = MultiForm.t(description="form data"),
        cookie: dict = Cookie.t(raw_return=True, description="cookie"),
        multi_user_name: List[str] = MultiQuery.t(description="user name", min_length=2, max_length=4),
        age: int = Path.t(description="age", gt=1, lt=100),
        uid: int = Query.t(description="user id", gt=10, lt=1000),
        user_name: str = Query.t(description="user name", min_length=2, max_length=4),
        email: Optional[str] = Query.t(default="example@xxx.com", description="user email"),
        sex: SexEnum = Query.t(description="sex"),
    ) -> None:
        self.write(
            {
                "code": 0,
                "msg": "",
                "data": {
                    "form_a": a,
                    "form_b": b,
                    "form_c": c,
                    "cookie": cookie,
                    "multi_user_name": multi_user_name,
                    "age": age,
                    "uid": uid,
                    "user_name": user_name,
                    "email": email,
                    "sex": sex,
                },
            }
        )


app: Application = Application([(r"/api/demo/(?P<age>\w+)", DemoHandler)])
AddDocRoute(app)


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

Note

为了确保演示的代码能够在不同的机器上顺利运行，这里没有演示File字段的用法，具体用法请参考不同Web框架示例代码中的field_route.py文件中/api/pait-base-field对应的路由函数。

示例代码演示了通过不同种类Field从请求对象获取请求者的参数，并组装成一定的格式返回。 接下来运行示例代码，然后使用curl命令在终端进行一次请求测试，命令如下:

curl -X 'POST' \
  'http://127.0.0.1:8000/api/demo/18?multi_user_name=aaa&multi_user_name=bbb&uid=999&user_name=so1n&sex=man' \
  -H 'accept: */*' \
  -H 'Cookie: cookie=aaa,aaa' \
  -H 'Content-Type: application/x-www-form-urlencoded' \
  -d 'a=aaa&b=bbb&c=ccc1&c=ccc2'

正常情况下，会在终端看到如下输出:

{
    "code": 0,
    "data": {
        "age": 18,
        "cookie": {
            "cookie": "aaa,aaa"
        },
        "email": "example@xxx.com",
        "form_a": "aaa",
        "form_b": "bbb",
        "form_c": [
            "ccc1",
            "ccc2"
        ],
        "multi_user_name": [
            "aaa",
            "bbb"
        ],
        "sex": "man",
        "uid": 999,
        "user_name": "so1n"
    },
    "msg": ""
}

通过输出结果可以发现，Pait都能通过Field的种类准确的从请求对象获取对应的值。

2.Field的功能

从上面的例子可以看到url没有携带email参数， 但是接口返回的响应值中的email却为example@xxx.com。 这是因为email字段的Field的default属性被设置为example@xx.com， 这样Pait会在无法通过请求体获取到email值的的情况下，把默认值赋给变量。

除了默认值之外， Field也有很多的功能，这些功能大部分来源于Field所继承的pydantic.Field。

2.1.default

Pait通过读取Field的default属性来获取该参数的默认值，当Field的default属性不为空且请求体没有对应的值时，Pait就会把default的值注入到对应的变量中。

下面是简单的示例代码，示例代码中的两个接口都直接返回获取到的值demo_value，其中demo接口带有默认值， 默认值为字符串123，而demo1接口没有默认值:

FlaskStarletteSanicTornado

from flask import Flask

from pait import field
from pait.app.flask import pait
from pait.exceptions import TipException


def api_exception(exc: Exception) -> str:
    if isinstance(exc, TipException):
        exc = exc.exc
    return str(exc)


@pait()
def demo(demo_value: str = field.Query.t(default="123")) -> str:
    return demo_value


@pait()
def demo1(demo_value: str = field.Query.t()) -> str:
    return demo_value


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


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

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

from pait import field
from pait.app.starlette import pait
from pait.exceptions import TipException


async def api_exception(request: Request, exc: Exception) -> PlainTextResponse:
    if isinstance(exc, TipException):
        exc = exc.exc
    return PlainTextResponse(str(exc))


@pait()
async def demo(demo_value: str = field.Query.t(default="123")) -> PlainTextResponse:
    return PlainTextResponse(demo_value)


@pait()
async def demo1(demo_value: str = field.Query.t()) -> PlainTextResponse:
    return PlainTextResponse(demo_value)


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

from sanic import HTTPResponse, Request, Sanic

from pait import field
from pait.app.sanic import pait
from pait.exceptions import TipException


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


@pait()
async def demo(demo_value: str = field.Query.t(default="123")) -> HTTPResponse:
    return HTTPResponse(demo_value)


@pait()
async def demo1(demo_value: str = field.Query.t()) -> HTTPResponse:
    return HTTPResponse(demo_value)


app = Sanic("demo")
app.add_route(demo, "/api/demo", methods={"GET"})
app.add_route(demo1, "/api/demo1", methods={"GET"})
app.exception(Exception)(api_exception)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

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.openapi.doc_route import AddDocRoute


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()
    async def get(self, demo_value: str = field.Query.t(default="123")) -> None:
        self.write(demo_value)


class Demo1Handler(_Handler):
    @pait()
    async def get(self, demo_value: str = field.Query.t()) -> None:
        self.write(demo_value)


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


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

在运行代码且调用curl后可以发现，当没有传demo_value参数时，/api/demo路由默认返回123, 而/api/demo1路由会抛出找不到demo_value值的错误，如下:

curl "http://127.0.0.1:8000/api/demo"123curl "http://127.0.0.1:8000/api/demo1"Can not found demo_value value

当传递的demo_value参数为456时，/api/demo和/api/demo1路由都会返回456:

curl "http://127.0.0.1:8000/api/demo?demo_value=456"456curl "http://127.0.0.1:8000/api/demo1?demo_value=456"456

Note

错误处理使用了TipException，可以通过异常提示了解TipException的作用。

2.2.default_factory

default_factory的作用与default类似，只不过default_factory接收的值是函数，只有当请求命中路由函数且Pait无法从请求对象中找到变量需要的值时才会被执行并将返回值注入到变量中。

示例代码如下，第一个接口的默认值是当前时间， 第二个接口的默认值是uuid，他们每次的返回值都是收到请求时生成的:

FlaskStarletteSanicTornado

import datetime
import uuid

from flask import Flask

from pait import field
from pait.app.flask import pait
from pait.exceptions import TipException


def api_exception(exc: Exception) -> str:
    if isinstance(exc, TipException):
        exc = exc.exc
    return str(exc)


@pait()
def demo(demo_value: datetime.datetime = field.Query.t(default_factory=datetime.datetime.now)) -> str:
    return str(demo_value)


@pait()
def demo1(demo_value: str = field.Query.t(default_factory=lambda: uuid.uuid4().hex)) -> str:
    return demo_value


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


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

import datetime
import uuid

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

from pait import field
from pait.app.starlette import pait
from pait.exceptions import TipException


async def api_exception(request: Request, exc: Exception) -> PlainTextResponse:
    if isinstance(exc, TipException):
        exc = exc.exc
    return PlainTextResponse(str(exc))


@pait()
async def demo(
    demo_value: datetime.datetime = field.Query.t(default_factory=datetime.datetime.now),
) -> PlainTextResponse:
    return PlainTextResponse(str(demo_value))


@pait()
async def demo1(demo_value: str = field.Query.t(default_factory=lambda: uuid.uuid4().hex)) -> PlainTextResponse:
    return PlainTextResponse(demo_value)


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

import datetime
import uuid

from sanic import HTTPResponse, Request, Sanic

from pait import field
from pait.app.sanic import pait
from pait.exceptions import TipException


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


@pait()
async def demo(demo_value: datetime.datetime = field.Query.t(default_factory=datetime.datetime.now)) -> HTTPResponse:
    return HTTPResponse(str(demo_value))


@pait()
async def demo1(demo_value: str = field.Query.t(default_factory=lambda: uuid.uuid4().hex)) -> HTTPResponse:
    return HTTPResponse(demo_value)


app = Sanic("demo")
app.add_route(demo, "/api/demo", methods={"GET"})
app.add_route(demo1, "/api/demo1", methods={"GET"})
app.exception(Exception)(api_exception)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

import datetime
import uuid

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.openapi.doc_route import AddDocRoute


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()
    async def get(self, demo_value: datetime.datetime = field.Query.t(default_factory=datetime.datetime.now)) -> None:
        self.write(str(demo_value))


class Demo1Handler(_Handler):
    @pait()
    async def get(self, demo_value: str = field.Query.t(default_factory=lambda: uuid.uuid4().hex)) -> None:
        self.write(demo_value)


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


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

在运行代码并使用curl调用可以发现接口每次返回的结果都是不一样的:

curl "http://127.0.0.1:8000/api/demo"2022-02-07T14:54:29.127519curl "http://127.0.0.1:8000/api/demo"2022-02-07T14:54:33.789994curl "http://127.0.0.1:8000/api/demo1"7e4659e18103471da9db91ed4843d962curl "http://127.0.0.1:8000/api/demo1"ef84f04fa9fc4ea9a8b44449c76146b8

2.3.alias

通常情况下Pait会以参数名为key从请求体中获取数据，但有一些参数名如Content-Type是Python不支持的变量命名方式， 此时可以使用alias来为变量设置别名，如下示例代码:

FlaskStarletteSanicTornado

from flask import Flask

from pait import field
from pait.app.flask import pait


@pait()
def demo(content_type: str = field.Header.t(alias="Content-Type")) -> str:
    return content_type


app = Flask("demo")

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


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

from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
from starlette.routing import Route

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


@pait()
async def demo(content_type: str = field.Header.t(alias="Content-Type")) -> PlainTextResponse:
    return PlainTextResponse(content_type)


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

from sanic import HTTPResponse, Sanic

from pait import field
from pait.app.sanic import pait


@pait()
async def demo(content_type: str = field.Header.t(alias="Content-Type")) -> HTTPResponse:
    return HTTPResponse(content_type)


app = Sanic("demo")
app.add_route(demo, "/api/demo", methods={"GET"})


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

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

from pait import field
from pait.app.tornado import pait
from pait.openapi.doc_route import AddDocRoute


class DemoHandler(RequestHandler):
    @pait()
    async def get(self, content_type: str = field.Header.t(alias="Content-Type")) -> None:
        self.write(content_type)


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


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

运行代码并调用curl命令后可以发现，Pait正常的从请求体的Header中提取Content-Type的值并赋给了content_type变量，所以路由函数能正常返回值123:

curl "http://127.0.0.1:8000/api/demo" -H "Content-Type:123"123

2.4.数值类型校验之gt，ge，lt，le，multiple_of

gt，ge，lt，le，multiple_of都属于pydantic的数值类型校验， 仅用于数值的类型，他们的作用各不相同：

• gt：仅用于数值的类型，会校验数值是否大于该值，同时也会在OpenAPI添加exclusiveMinimum属性。
• ge：仅用于数值的类型，会校验数值是否大于等于该值，同时也会在OpenAPI添加exclusiveMinimum属性。
• lt：仅用于数值的类型，会校验数值是否小于该值，同时也会在OpenAPI添加exclusiveMaximum属性。
• le：仅用于数值的类型，会校验数值是否小于等于该值，同时也会在OpenAPI添加exclusiveMaximum属性。
• multiple_of：仅用于数字， 会校验该数字是否是指定值得倍数。

使用方法如下:

FlaskStarletteSanicTornado

from flask import Flask, Response, jsonify
from pydantic import ValidationError

from pait import field
from pait.app.flask import pait
from pait.exceptions import TipException


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


@pait()
def demo(
    demo_value1: int = field.Query.i(gt=1, lt=10),
    demo_value2: int = field.Query.i(ge=1, le=1),
    demo_value3: int = field.Query.i(multiple_of=3),
) -> dict:
    return {"data": [demo_value1, demo_value2, demo_value3]}


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)

from pydantic import ValidationError
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


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


@pait()
async def demo(
    demo_value1: int = field.Query.i(gt=1, lt=10),
    demo_value2: int = field.Query.i(ge=1, le=1),
    demo_value3: int = field.Query.i(multiple_of=3),
) -> JSONResponse:
    return JSONResponse({"data": [demo_value1, demo_value2, demo_value3]})


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

from pydantic import ValidationError
from sanic import HTTPResponse, Request, Sanic, json

from pait import field
from pait.app.sanic import pait
from pait.exceptions import TipException


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


@pait()
async def demo(
    demo_value1: int = field.Query.i(gt=1, lt=10),
    demo_value2: int = field.Query.i(ge=1, le=1),
    demo_value3: int = field.Query.i(multiple_of=3),
) -> HTTPResponse:
    return json({"data": [demo_value1, demo_value2, demo_value3]})


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

from pydantic import ValidationError
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.openapi.doc_route import AddDocRoute


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

        if isinstance(exc, ValidationError):
            self.write({"data": exc.errors()})
        else:
            self.write({"data": str(exc)})
        self.finish()


class DemoHandler(_Handler):
    @pait()
    async def get(
        self,
        demo_value1: int = field.Query.i(gt=1, lt=10),
        demo_value2: int = field.Query.i(ge=1, le=1),
        demo_value3: int = field.Query.i(multiple_of=3),
    ) -> None:
        self.write({"data": [demo_value1, demo_value2, demo_value3]})


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


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

这份示例代码只有一个接口，但是接受了三个参数demo_value1, demo_value2, demo_value3，他们分别只接收符合大于1小于10，等于1以及3的倍数的三个参数。

在运行代码并调用curl命令可以发现第一个请求符合要求并得到了正确的响应结果， 第二，三，四个请求分别是demo_value1，demo_value2，demo_value3的值不在要求的范围内，所以Pait会生成Pydantic.ValidationError的错误信息，从错误信息可以简单的看出来三个参数都不符合接口设置的限定条件：

➜  ~ curl "http://127.0.0.1:8000/api/demo?demo_value1=2&demo_value2=1&demo_value3=3"
{"data":[2,1,3]}
➜  ~ curl "http://127.0.0.1:8000/api/demo?demo_value1=11&demo_value2=1&demo_value3=3"
{
    "data": [
        {
            "ctx": {"limit_value": 10},
            "loc": ["query", "demo_value1"],
            "msg": "ensure this value is less than 10",
            "type": "value_error.number.not_lt"
        }
    ]
}
➜  ~ curl "http://127.0.0.1:8000/api/demo?demo_value1=2&demo_value2=2&demo_value3=3"
{
    "data": [
        {
            "ctx": {"limit_value": 1},
            "loc": ["query", "demo_value2"],
            "msg": "ensure this value is less than or equal to 1",
            "type": "value_error.number.not_le"
        }
    ]
}
➜  ~ curl "http://127.0.0.1:8000/api/demo?demo_value1=2&demo_value2=1&demo_value3=4"
{
    "data": [
        {
            "ctx": {"multiple_of": 3},
            "loc": ["query", "demo_value3"],
            "msg": "ensure this value is a multiple of 3",
            "type": "value_error.number.not_multiple"
        }
    ]
}

2.5.数组校验之min_items，max_items

min_items，max_items都属于pydantic的Sequence类型校验，仅用于Sequence的类型，他们的作用各不相同：

• min_items：仅用于Sequence类型，会校验Sequence长度是否满足大于等于指定的值。
• max_items： 仅用于Sequence类型，会校验Sequence长度是否满足小于等于指定的值。

Note

如果使用的Pydantic版本大于2.0.0，请使用min_length和max_length代替min_items和max_items。

示例代码如下，该路由函数通过field.MultiQuery从请求Url中获取参数demo_value的数组，并返回给调用端，其中数组的长度限定在1到2之间：

FlaskStarletteSanicTornado

from typing import List

from flask import Flask, Response, jsonify
from pydantic import ValidationError

from pait import field
from pait.app.flask import pait
from pait.exceptions import TipException


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


@pait()
def demo(
    demo_value: List[int] = field.MultiQuery.i(min_items=1, max_items=2),
) -> dict:
    return {"data": demo_value}


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)

from typing import List

from pydantic import ValidationError
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


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


@pait()
async def demo(
    demo_value: List[int] = field.MultiQuery.i(min_items=1, max_items=2),
) -> JSONResponse:
    return JSONResponse({"data": demo_value})


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

from typing import List

from pydantic import ValidationError
from sanic import HTTPResponse, Request, Sanic, json

from pait import field
from pait.app.sanic import pait
from pait.exceptions import TipException


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


@pait()
async def demo(demo_value: List[int] = field.MultiQuery.i(min_items=1, max_items=2)) -> HTTPResponse:
    return json({"data": demo_value})


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

from typing import List

from pydantic import ValidationError
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.openapi.doc_route import AddDocRoute


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

        if isinstance(exc, ValidationError):
            self.write({"data": exc.errors()})
        else:
            self.write({"data": str(exc)})
        self.finish()


class DemoHandler(_Handler):
    @pait()
    async def get(self, demo_value: List[int] = field.MultiQuery.i(min_items=1, max_items=2)) -> None:
        self.write({"data": demo_value})


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


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

与2.4一样，通过curl调用可以发现合法的参数会放行，不合法的参数则会抛错：

➜  ~ curl "http://127.0.0.1:8000/api/demo?demo_value=1"
{"data":[1]}
➜  ~ curl "http://127.0.0.1:8000/api/demo?demo_value=1&demo_value=2"
{"data":[1,2]}
➜  ~ curl "http://127.0.0.1:8000/api/demo?demo_value=1&demo_value=2&demo_value=3"
{
    "data": [
        {
            "loc": [
                "demo_value"
            ],
            "msg": "ensure this value has at most 2 items",
            "type": "value_error.list.max_items",
            "ctx": {
                "limit_value": 2
            }
        }
    ]
}

2.6.字符串校验之min_length，max_length，regex

min_length，max_length，regex都属于pydantic的字符串类型校验，仅用于字符串的类型，他们的作用各不相同：

• min_length：仅用于字符串类型，会校验字符串的长度是否满足大于等于指定的值。
• max_length：仅用于字符串类型，会校验字符串的长度是否满足小于等于指定的值。
• regex：仅用于字符串类型，会校验字符串是否符合该正则表达式。

Note

如果使用的Pydantic版本大于2.0.0，min_length和max_length还可以校验序列类型，并考虑使用pattern代替regex。

示例代码如下， 该路由函数需要从Url中获取一个长度大小为6并以英文字母u开头的值：

FlaskStarletteSanicTornado

from flask import Flask, Response, jsonify
from pydantic import ValidationError

from pait import field
from pait.app.flask import pait
from pait.exceptions import TipException


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


@pait()
def demo(demo_value: str = field.Query.i(min_length=6, max_length=6, regex="^u")) -> dict:
    return {"data": demo_value}


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)

from pydantic import ValidationError
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


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


@pait()
async def demo(demo_value: str = field.Query.i(min_length=6, max_length=6, regex="^u")) -> JSONResponse:
    return JSONResponse({"data": demo_value})


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

from pydantic import ValidationError
from sanic import HTTPResponse, Request, Sanic, json

from pait import field
from pait.app.sanic import pait
from pait.exceptions import TipException


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


@pait()
async def demo(demo_value: str = field.Query.i(min_length=6, max_length=6, regex="^u")) -> HTTPResponse:
    return json({"data": demo_value})


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

from pydantic import ValidationError
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.openapi.doc_route import AddDocRoute


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

        if isinstance(exc, ValidationError):
            self.write({"data": exc.errors()})
        else:
            self.write({"data": str(exc)})
        self.finish()


class DemoHandler(_Handler):
    @pait()
    async def get(self, demo_value: str = field.Query.i(min_length=6, max_length=6, regex="^u")) -> None:
        self.write({"data": demo_value})


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


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

运行代码并使用curl进行三次请求，通过结果可以看出，第一次请求的结果为正常，第二次请求的结果不符合正则表达式，而第三次请求的结果长度不符合要求：

➜  ~ curl "http://127.0.0.1:8000/api/demo?demo_value=u66666"
{"data":"u66666"}
➜  ~ curl "http://127.0.0.1:8000/api/demo?demo_value=666666"
{"data":[{"loc":["demo_value"],"msg":"string does not match regex \"^u\"","type":"value_error.str.regex","ctx":{"pattern":"^u"}}]}
➜  ~ curl "http://127.0.0.1:8000/api/demo?demo_value=1"
{"data":[{"loc":["demo_value"],"msg":"ensure this value has at least 6 characters","type":"value_error.any_str.min_length","ctx":{"limit_value":6}}]}

2.7.raw_return

该参数的默认值为False，如果为True，则Pait不会根据参数名或者alias为key从请求数据获取值， 而是把整个请求值返回给对应的变量。

示例代码如下， 该接口为一个POST接口， 它需要两个值，第一个值为整个客户端传过来的Json参数， 而第二个值为客户端传过来的Json参数中Key为a的值：

FlaskStarletteSanicTornado

from flask import Flask, Response, jsonify
from pydantic import ValidationError

from pait import field
from pait.app.flask import pait
from pait.exceptions import TipException


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


@pait()
def demo(
    demo_value: dict = field.Json.i(raw_return=True),
    a: str = field.Json.i(),
) -> dict:
    return {"data": demo_value, "a": a}


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


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

from pydantic import ValidationError
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


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


@pait()
async def demo(
    demo_value: dict = field.Json.i(raw_return=True),
    a: str = field.Json.i(),
) -> JSONResponse:
    return JSONResponse({"data": demo_value, "a": a})


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

from pydantic import ValidationError
from sanic import HTTPResponse, Request, Sanic, json

from pait import field
from pait.app.sanic import pait
from pait.exceptions import TipException


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


@pait()
async def demo(
    demo_value: dict = field.Json.i(raw_return=True),
    a: str = field.Json.i(),
) -> HTTPResponse:
    return json({"data": demo_value, "a": a})


app = Sanic("demo")
app.add_route(demo, "/api/demo", methods={"POST"})
app.exception(Exception)(api_exception)


if __name__ == "__main__":
    import uvicorn  # type: ignore

    uvicorn.run(app)

from pydantic import ValidationError
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.openapi.doc_route import AddDocRoute


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

        if isinstance(exc, ValidationError):
            self.write({"data": exc.errors()})
        else:
            self.write({"data": str(exc)})
        self.finish()


class DemoHandler(_Handler):
    @pait()
    async def post(
        self,
        demo_value: dict = field.Json.i(raw_return=True),
        a: str = field.Json.i(),
    ) -> None:
        self.write({"data": demo_value, "a": a})


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


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

运行代码并使用curl调用， 可以发现结果符合预期：

curl "http://127.0.0.1:8000/api/demo" \ -X POST -d '{"a": "1", "b": "2"}' \ --header "Content-Type: application/json"
{"demo_value":{"a":"1","b":"2"},"a":"1"}

2.8.自定义查询不到值的异常

在正常情况下，如果请求对象中没有Pait需要的数据，那么Pait会抛出NotFoundValueException异常。 除此之外，Pait还支持开发者通过not_value_exception_func自定义异常处理， 如下代码中路由函数有两个变量，第一个变量demo_value1没有设置任何Field的属性， 而第二个变量demo_value2设置了not_value_exception_func属性为lambda param: RuntimeError(f"not found {param.name} data")：

FlaskStarletteSanicTornado

from flask import Flask, Response, jsonify
from pydantic import ValidationError

from pait import field
from pait.app.flask import pait
from pait.exceptions import TipException


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


@pait()
def demo(
    demo_value1: str = field.Query.i(),
    demo_value2: str = field.Query.i(
        not_value_exception_func=lambda param: RuntimeError(f"not found {param.name} data")
    ),
) -> dict:
    return {"data": {"demo_value1": demo_value1, "demo_value2": demo_value2}}


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)

from pydantic import ValidationError
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


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


@pait()
async def demo(
    demo_value1: str = field.Query.i(),
    demo_value2: str = field.Query.i(
        not_value_exception_func=lambda param: RuntimeError(f"not found {param.name} data")
    ),
) -> JSONResponse:
    return JSONResponse({"data": {"demo_value1": demo_value1, "demo_value2": demo_value2}})


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

from pydantic import ValidationError
from sanic import HTTPResponse, Request, Sanic, json

from pait import field
from pait.app.sanic import pait
from pait.exceptions import TipException


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


@pait()
async def demo(
    demo_value1: str = field.Query.i(),
    demo_value2: str = field.Query.i(
        not_value_exception_func=lambda param: RuntimeError(f"not found {param.name} data")
    ),
) -> HTTPResponse:
    return json({"data": {"demo_value1": demo_value1, "demo_value2": demo_value2}})


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

from pydantic import ValidationError
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.openapi.doc_route import AddDocRoute


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

        if isinstance(exc, ValidationError):
            self.write({"data": exc.errors()})
        else:
            self.write({"data": str(exc)})
        self.finish()


class DemoHandler(_Handler):
    @pait()
    async def get(
        self,
        demo_value1: str = field.Query.i(),
        demo_value2: str = field.Query.i(
            not_value_exception_func=lambda param: RuntimeError(f"not found {param.name} data")
        ),
    ) -> None:
        self.write({"data": {"demo_value1": demo_value1, "demo_value2": demo_value2}})


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


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

接着运行代码，并在终端执行如下curl命令：

curl "http://127.0.0.1:8000/api/demo?demo_value1=1&demo_value2=2"{"data": {"demo_value1": "1", "demo_value2": "2"}}curl "http://127.0.0.1:8000/api/demo?demo_value2=2"{"data": "Can not found demo_value1 value"}curl "http://127.0.0.1:8000/api/demo?demo_value1=1"{"data":"not found demo_value2 data"}

通过输出结果可以看到demo_value1缺值和demo_value2缺值的响应是不同的，其中demo_value2的缺值异常消息由lambda param: RuntimeError(f"not found {param.name} data")抛出的。

2.8.其它功能

除了上述功能外，Pait还有其它功能，可以到对应模块文档了解：

属性	文档	描述
links	OpenAPI	用于支持OpenAPI的link功能
media_type	OpenAPI	Field对应的media_type，用于OpenAPI Scheme的media type。
openapi_serialization	OpenAPI	指定OpenAPI Schema的序列化方式。
example	OpenAPI	用于文档的示例值,Mock请求与响应等Mock功能，参数值支持变量和可调用函数如datetime.datetim.now，推荐与faker一起使用。
description	OpenAPI	用于OpenAPI的参数描述
openapi_include	OpenAPI	定义该字段不被OpenAPI读取
extra_param_list	Plugin	定义插件与参数之间的关系