跳转至

如何使用Type

变量的类型--Type用于指明该变量值的最终类型是什么，通过定义Type可以拓展变量的校验规则，如下示例代码:

from pait.app.any import pait
from pait import field

@pait()
def demo(
    a: str = field.Body.i(),
    b: int = field.Body.i(),
) -> dict:
    return {"a": a, "b": b}

在代码运行时，Pait会在内部将函数签名转换为如下的Pydantic.BaseModel:

from pydantic import BaseModel, Field

class Demo(BaseModel):
    a: str = Field()
    b: int = Field()

所以路由函数中的Type可以变得非常灵活，使用者可以像Pydantic Field Types一样使用Type，此外，还可以通过编写符合Pait规范的Type来使用Pait的一些功能。

1.Type的值为Pydantic.BaseModel

在经历了一段时间的开发后，会发现有些路由函数的参数是可以复用的，这时可以采用Type的值为Pydantic.BaseModel的方案把路由函数的参数转化为pydantic.Basemodel。

此外，还有两个不一样的路由函数demo和demo1。 其中demo会从Url Path中获取所有的值并交给DemoModel进行校验再返回.dict方法生成的数据。 而demo1虽然与路由函数demo很像， 只不过获取数据的来源从Url Path变为Json Body。

docs_source_code/docs_source_code/introduction/how_to_use_type/flask_with_model_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 flask import Flask, Response, jsonify
from pydantic import BaseModel, Field

from pait import _pydanitc_adapter, field
from pait.app.flask import pait

if _pydanitc_adapter.is_v1:

    class DemoModel(BaseModel):
        uid: str = Field(max_length=6, min_length=6, regex="^u")
        name: str = Field(min_length=4, max_length=10)
        age: int = Field(ge=0, le=100)

else:

    class DemoModel(BaseModel):  # type: ignore
        uid: str = Field(max_length=6, min_length=6, pattern="^u")
        name: str = Field(min_length=4, max_length=10)
        age: int = Field(ge=0, le=100)


@pait()
def demo(demo_model: DemoModel = field.Query.i(raw_return=True)) -> Response:
    return jsonify(demo_model.dict())


@pait()
def demo1(demo_model: DemoModel = field.Json.i(raw_return=True)) -> Response:
    return jsonify(demo_model.dict())


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


if __name__ == "__main__":
    app.run(port=8000)
docs_source_code/docs_source_code/introduction/how_to_use_type/starlette_with_model_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 pydantic import BaseModel, Field
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

from pait import _pydanitc_adapter, field
from pait.app.starlette import pait

if _pydanitc_adapter.is_v1:

    class DemoModel(BaseModel):
        uid: str = Field(max_length=6, min_length=6, regex="^u")
        name: str = Field(min_length=4, max_length=10)
        age: int = Field(ge=0, le=100)

else:

    class DemoModel(BaseModel):  # type: ignore
        uid: str = Field(max_length=6, min_length=6, pattern="^u")
        name: str = Field(min_length=4, max_length=10)
        age: int = Field(ge=0, le=100)


@pait()
async def demo(demo_model: DemoModel = field.Query.i(raw_return=True)) -> JSONResponse:
    return JSONResponse(demo_model.dict())


@pait()
async def demo1(demo_model: DemoModel = field.Json.i(raw_return=True)) -> JSONResponse:
    return JSONResponse(demo_model.dict())


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)
docs_source_code/docs_source_code/introduction/how_to_use_type/sanic_with_model_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 pydantic import BaseModel, Field
from sanic import Sanic, json, response

from pait import _pydanitc_adapter, field
from pait.app.sanic import pait

if _pydanitc_adapter.is_v1:

    class DemoModel(BaseModel):
        uid: str = Field(max_length=6, min_length=6, regex="^u")
        name: str = Field(min_length=4, max_length=10)
        age: int = Field(ge=0, le=100)

else:

    class DemoModel(BaseModel):  # type: ignore
        uid: str = Field(max_length=6, min_length=6, pattern="^u")
        name: str = Field(min_length=4, max_length=10)
        age: int = Field(ge=0, le=100)


@pait()
async def demo(demo_model: DemoModel = field.Query.i(raw_return=True)) -> response.HTTPResponse:
    return json(demo_model.dict())


@pait()
async def demo1(demo_model: DemoModel = field.Json.i(raw_return=True)) -> response.HTTPResponse:
    return json(demo_model.dict())


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)
docs_source_code/docs_source_code/introduction/how_to_use_type/tornado_with_model_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 pydantic import BaseModel, Field
from tornado.ioloop import IOLoop
from tornado.web import Application, RequestHandler

from pait import _pydanitc_adapter, field
from pait.app.tornado import pait

if _pydanitc_adapter.is_v1:

    class DemoModel(BaseModel):
        uid: str = Field(max_length=6, min_length=6, regex="^u")
        name: str = Field(min_length=4, max_length=10)
        age: int = Field(ge=0, le=100)

else:

    class DemoModel(BaseModel):  # type: ignore
        uid: str = Field(max_length=6, min_length=6, pattern="^u")
        name: str = Field(min_length=4, max_length=10)
        age: int = Field(ge=0, le=100)


class DemoHandler(RequestHandler):
    @pait()
    def get(self, demo_model: DemoModel = field.Query.i(raw_return=True)) -> None:
        self.write(demo_model.dict())


class Demo1Handler(RequestHandler):
    @pait()
    def post(self, demo_model: DemoModel = field.Json.i(raw_return=True)) -> None:
        self.write(demo_model.dict())


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


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

Note

如果Field对象的raw_return不为True，则Pait会以Key为demo_model从请求资源中获取值

接下来运行代码并使用curl对两个接口进行测试：

➜  ~ curl "http://127.0.0.1:8000/api/demo?uid=u12345&name=so1n&age=10"
{"uid":"u12345","name":"so1n","age":10}
➜  ~ curl "http://127.0.0.1:8000/api/demo1" -X POST -d '{"uid": "u12345", "name": "so1n", "age": 10}' --header "Content-Type: application/json"
{"uid":"u12345","name":"so1n","age":10}

通过输出的结果可以发现两个路由函数都能正常的工作，但是在这种用法下BaseModel的所有字段都只能使用同一种Field类型，这时可以采用另外一种方法。

2.Type的值为特殊的Pydantic.BaseModel

Pait的Field对象是一个携带资源来源标识的pydantic.FieldInfo对象，同时也支持转为标准的pydantic.FieldInfo方法， 所以可以用于pydantic.BaseModel中，比如前文说到的DemoModel对象可以改写为如下代码:

from pait import field

from pydantic import BaseModel

class DemoModel(BaseModel):
    uid: str = field.Query.i(max_length=6, min_length=6, regex="^u")
    name: str = field.Body.i(min_length=4, max_length=10)
    age: int = field.Header.i(ge=0, le=100)

可以看到DemoModel每个属性的Field对象都是不一样的，不过为了能让Pait正确的加载DemoModel填写的参数形式需要从<name>:<type>=<default>改为<name>:<type>，如下:

@pait()
def demo(demo_model: DemoModel) -> None:
    pass

2.1.DefaultField

除此之外，还可以使用Pait的DefaultField功能，该功能可以根据路由函数定义的Field来自动替换pydantic.BaseModel中每个属性的Field，如下还是一个普通的DemoModel:

from pydantic import BaseModel, Field

class DemoModel(BaseModel):
    uid: str = Field(max_length=6, min_length=6, regex="^u")
    name: str = Field(min_length=4, max_length=10)
    age: int = Field(ge=0, le=100)

然后通过pait装饰器中的default_field_class属性指定了demo路由的DefaultField为Query，demo1路由的DefaultField为Body:

from pait import field
from pait.app.any import pait


@pait(default_field_class=field.Query)
def demo(demo_model: DemoModel) -> None:
    pass

@pait(default_field_class=field.Body)
def demo1(demo_model: DemoModel) -> None:
    pass

这样一来，就可以在使用不同请求资源的路由函数中使用同一个DemoModel，完整的代码如下:

docs_source_code/docs_source_code/introduction/how_to_use_type/flask_with_pait_model_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
41
42
43
44
from uuid import uuid4

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

from pait import _pydanitc_adapter, field
from pait.app.flask import pait

if _pydanitc_adapter.is_v1:

    class DemoModel(BaseModel):
        uid: str = Field(..., max_length=6, min_length=6, regex="^u")
        name: str = Field(..., min_length=4, max_length=10)
        age: int = Field(..., ge=0, le=100)

        request_id: str = field.Header.i(default_factory=lambda: str(uuid4()))

else:

    class DemoModel(BaseModel):  # type: ignore
        uid: str = Field(..., max_length=6, min_length=6, pattern="^u")
        name: str = Field(..., min_length=4, max_length=10)
        age: int = Field(..., ge=0, le=100)

        request_id: str = field.Header.i(default_factory=lambda: str(uuid4()))


@pait(default_field_class=field.Query)
def demo(demo_model: DemoModel) -> Response:
    return jsonify(demo_model.dict())


@pait(default_field_class=field.Body)
def demo1(demo_model: DemoModel) -> Response:
    return jsonify(demo_model.dict())


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


if __name__ == "__main__":
    app.run(port=8000)
docs_source_code/docs_source_code/introduction/how_to_use_type/starlette_with_pait_model_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
41
42
43
44
45
46
from uuid import uuid4

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

from pait import _pydanitc_adapter, field
from pait.app.starlette import pait

if _pydanitc_adapter.is_v1:

    class DemoModel(BaseModel):
        uid: str = Field(..., max_length=6, min_length=6, regex="^u")
        name: str = Field(..., min_length=4, max_length=10)
        age: int = Field(..., ge=0, le=100)

        request_id: str = field.Header.i(default_factory=lambda: str(uuid4()))

else:

    class DemoModel(BaseModel):  # type: ignore
        uid: str = Field(..., max_length=6, min_length=6, pattern="^u")
        name: str = Field(..., min_length=4, max_length=10)
        age: int = Field(..., ge=0, le=100)

        request_id: str = field.Header.i(default_factory=lambda: str(uuid4()))


@pait(default_field_class=field.Query)
async def demo(demo_model: DemoModel) -> JSONResponse:
    return JSONResponse(demo_model.dict())


@pait(default_field_class=field.Body)
async def demo1(demo_model: DemoModel) -> JSONResponse:
    return JSONResponse(demo_model.dict())


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)
docs_source_code/docs_source_code/introduction/how_to_use_type/sanic_with_pait_model_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
41
42
43
44
45
46
from uuid import uuid4

from pydantic import BaseModel, Field
from sanic import Sanic, json, response

from pait import _pydanitc_adapter, field
from pait.app.sanic import pait

if _pydanitc_adapter.is_v1:

    class DemoModel(BaseModel):
        uid: str = Field(..., max_length=6, min_length=6, regex="^u")
        name: str = Field(..., min_length=4, max_length=10)
        age: int = Field(..., ge=0, le=100)

        request_id: str = field.Header.i(default_factory=lambda: str(uuid4()))

else:

    class DemoModel(BaseModel):  # type: ignore
        uid: str = Field(..., max_length=6, min_length=6, pattern="^u")
        name: str = Field(..., min_length=4, max_length=10)
        age: int = Field(..., ge=0, le=100)

        request_id: str = field.Header.i(default_factory=lambda: str(uuid4()))


@pait(default_field_class=field.Query)
async def demo(demo_model: DemoModel) -> response.HTTPResponse:
    return json(demo_model.dict())


@pait(default_field_class=field.Body)
async def demo1(demo_model: DemoModel) -> response.HTTPResponse:
    return json(demo_model.dict())


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)
docs_source_code/docs_source_code/introduction/how_to_use_type/tornado_with_pait_model_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
41
42
43
44
45
46
from uuid import uuid4

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

from pait import _pydanitc_adapter, field
from pait.app.tornado import pait

if _pydanitc_adapter.is_v1:

    class DemoModel(BaseModel):
        uid: str = Field(..., max_length=6, min_length=6, regex="^u")
        name: str = Field(..., min_length=4, max_length=10)
        age: int = Field(..., ge=0, le=100)

        request_id: str = field.Header.i(default_factory=lambda: str(uuid4()))

else:

    class DemoModel(BaseModel):  # type: ignore
        uid: str = Field(..., max_length=6, min_length=6, pattern="^u")
        name: str = Field(..., min_length=4, max_length=10)
        age: int = Field(..., ge=0, le=100)

        request_id: str = field.Header.i(default_factory=lambda: str(uuid4()))


class DemoHandler(RequestHandler):
    @pait(default_field_class=field.Query)
    def get(self, demo_model: DemoModel) -> None:
        self.write(demo_model.dict())


class Demo1Handler(RequestHandler):
    @pait(default_field_class=field.Body)
    def post(self, demo_model: DemoModel) -> None:
        self.write(demo_model.dict())


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


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

代码中的DemoModel增加了一个名为request_id的属性，它使用的Field对象是Header对象。 由于Default Field功能只会替换pydantic.FieldInfo，这意味着在运行的过程中request_id属性不会被Default Field影响，它还是会从Header获取数据。 现在运行代码，并调用curl命令可以看到如下输出：

➜  ~ curl "http://127.0.0.1:8000/api/demo?uid=u12345&name=so1n&age=10"
{"age":10,"name":"so1n","request_id":"6a5827f7-5767-46ef-91c6-f2ae99770865","uid":"u12345"}
➜  ~ curl "http://127.0.0.1:8000/api/demo1" -X POST -d '{"uid": "u12345", "name": "so1n", "age": 10}' --header "Content-Type: application/json"
{"age":10,"name":"so1n","request_id":"3279944c-6de7-4270-8536-33619641f25e","uid":"u12345"}

Note

需要注意的是在正常情况下，Pait是按照变量顺序去处理/校验每一个变量的值，如果发现有不合法的值就直接抛错，不会对剩下的值进行处理。 但是当Type的值为pydantic.BaseModel时，Pait会把参数委托给pydantic.BaseModel校验， 而pydantic.BaseModel会对所有值都进行校验，然后再把错误抛出来。

3.其它

3.1.Request对象

在使用Pait时，Request对象使用的频率会大幅的降低，所以Pait会自动把Request对象省略，比如原本Starlette框架路由函数的写法是：

from starlette.requests import Request


async def demo(request: Request) -> None:
    pass

而在使用了Pait后会变为如下代码：

from pait.app.starlette import pait


@pait()
async def demo():
    pass

Note

注：Sanic框架要求路由函数最少必须拥有一个参数。

如果需要在路由函数中使用Request对象，则需要以<var name>: <RequestType>的形式来定义路由函数参数，pait才能正确的把Request对象注入给对应的变量，例如：

docs_source_code/docs_source_code/introduction/how_to_use_type/flask_with_request_demo.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from flask import Flask, Request, Response, jsonify

from pait.app.flask import pait


@pait()
def demo(req: Request) -> Response:
    return jsonify({"url": req.path, "method": req.method})


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


if __name__ == "__main__":
    app.run(port=8000)
docs_source_code/docs_source_code/introduction/how_to_use_type/starlette_with_request_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
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

from pait.app.starlette import pait


@pait()
async def demo(req: Request) -> JSONResponse:
    return JSONResponse({"url": str(req.url.path), "method": req.method})


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)
docs_source_code/docs_source_code/introduction/how_to_use_type/sanic_with_request_demo.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
from sanic import Request, Sanic, json, response

from pait.app.sanic import pait


@pait()
async def demo(req: Request) -> response.HTTPResponse:
    return json({"url": req.path, "method": req.method})


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)
docs_source_code/docs_source_code/introduction/how_to_use_type/tornado_with_request_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
from tornado.httputil import HTTPServerRequest
from tornado.ioloop import IOLoop
from tornado.web import Application, RequestHandler

from pait.app.tornado import pait


class DemoHandler(RequestHandler):
    @pait()
    def get(self, req: HTTPServerRequest) -> None:
        self.write({"url": req.uri, "method": req.method})


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


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

运行代码并执行如下curl命令即可看到类似的结果：

curl "http://127.0.0.1:8000/api/demo"
{"url": "http://127.0.0.1:8000/api/demo", "method": "GET"}

3.2.如何自定义符合Pydantic要求且带有校验功能的Type

前文提到，在被Pait装饰的路由函数中使用的Type跟Pydantic的Type的作用是一样的，这也意味着可以通过Type来拓展校验规则从而弥补Field对象的不足， 比如在一个用户可能分布在不同国家的业务中，通常会使用时间戳来传递时间以防止时区不同带来的数据错误，如下：

docs_source_code/docs_source_code/introduction/how_to_use_type/flask_with_datetime_demo.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
import datetime

from flask import Flask, Response, jsonify

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


@pait()
def demo(timestamp: datetime.datetime = field.Query.i()) -> Response:
    return jsonify({"time": timestamp.isoformat()})


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


if __name__ == "__main__":
    app.run(port=8000)
docs_source_code/docs_source_code/introduction/how_to_use_type/starlette_with_datetime_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
import datetime

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

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


@pait()
async def demo(timestamp: datetime.datetime = field.Query.i()) -> JSONResponse:
    return JSONResponse({"time": timestamp.isoformat()})


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)
docs_source_code/docs_source_code/introduction/how_to_use_type/sanic_with_datetime_demo.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
import datetime

from sanic import Sanic, json, response

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


@pait()
async def demo(timestamp: datetime.datetime = field.Query.i()) -> response.HTTPResponse:
    return json({"time": timestamp.isoformat()})


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)
docs_source_code/docs_source_code/introduction/how_to_use_type/tornado_with_datetime_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
import datetime

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

from pait import field
from pait.app.tornado import pait


class DemoHandler(RequestHandler):
    @pait()
    def get(self, timestamp: datetime.datetime = field.Query.i()) -> None:
        self.write({"time": timestamp.isoformat()})


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


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

不过在运行代码并调用如下curl命令后可以发现，Pydantic自动把时间戳转为datetime类型且时区是UTC：

➜  ~ curl "http://127.0.0.1:8000/api/demo?timestamp=1600000000"
{"time":"2020-09-13T12:26:40+00:00"}

这种处理方式是没问题的，但假设这个业务的服务器是位于某个非UTC且数据库与程序的都依赖于机器的时区时， 就需要在每次获取数据后再进行一次时区转化。 为此，可以采用编写一个符合Pydantic校验规则的Type类来解决这个问题，实现代码如下:

import datetime
from typing import Callable, Generator, Union


class UnixDatetime(datetime.datetime):

    @classmethod
    def __get_validators__(cls) -> Generator[Callable, None, None]:
        yield cls.validate

    @classmethod
    def validate(cls, v: Union[int, str]) -> datetime.datetime:
        if isinstance(v, str):
            v = int(v)
        return datetime.datetime.fromtimestamp(v)

from pydantic import BeforeValidator
from typing import Union
from typing_extensions import Annotated
from datetime import datetime


def validate(v: Union[int, str]) -> datetime:
    if isinstance(v, str):
        v = int(v)
    return datetime.fromtimestamp(v)

UnixDatetime = Annotated[datetime, BeforeValidator(validate)]

通过示例代码可以看到由于Pydantic版本的不同，Type的实现也是不一样的，但是它们的逻辑是一样的，编写完成后就可以把Type应用到代码中：

docs_source_code/docs_source_code/introduction/how_to_use_type/flask_with_unix_datetime_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
41
42
43
44
45
import datetime
from typing import Callable, Generator, Union

from flask import Flask, Response, jsonify

from pait import field
from pait._pydanitc_adapter import is_v1
from pait.app.flask import pait

if is_v1:

    class UnixDatetime(datetime.datetime):
        @classmethod
        def __get_validators__(cls) -> Generator[Callable, None, None]:
            yield cls.validate

        @classmethod
        def validate(cls, v: Union[int, str]) -> datetime.datetime:
            if isinstance(v, str):
                v = int(v)
            return datetime.datetime.fromtimestamp(v)

else:
    from pydantic import BeforeValidator
    from typing_extensions import Annotated

    def validate(v: Union[int, str]) -> datetime.datetime:
        if isinstance(v, str):
            v = int(v)
        return datetime.datetime.fromtimestamp(v)

    UnixDatetime = Annotated[datetime.datetime, BeforeValidator(validate)]  # type: ignore


@pait()
def demo(timestamp: UnixDatetime = field.Query.i()) -> Response:
    return jsonify({"time": timestamp.isoformat()})


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


if __name__ == "__main__":
    app.run(port=8000)
docs_source_code/docs_source_code/introduction/how_to_use_type/starlette_with_unix_datetime_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
41
42
43
44
45
46
47
48
49
50
51
52
import datetime
from typing import Callable, Generator, Union

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

from pait import field
from pait._pydanitc_adapter import is_v1
from pait.app.starlette import pait

if is_v1:

    class UnixDatetime(datetime.datetime):
        @classmethod
        def __get_validators__(cls) -> Generator[Callable, None, None]:
            yield cls.validate

        @classmethod
        def validate(cls, v: Union[int, str]) -> datetime.datetime:
            if isinstance(v, str):
                v = int(v)
            return datetime.datetime.fromtimestamp(v)

else:
    from pydantic import BeforeValidator
    from typing_extensions import Annotated

    def validate(v: Union[int, str]) -> datetime.datetime:
        if isinstance(v, str):
            v = int(v)
        return datetime.datetime.fromtimestamp(v)

    UnixDatetime = Annotated[datetime.datetime, BeforeValidator(validate)]  # type: ignore


@pait()
async def demo(timestamp: UnixDatetime = field.Query.i()) -> JSONResponse:
    return JSONResponse({"time": timestamp.isoformat()})


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)
docs_source_code/docs_source_code/introduction/how_to_use_type/sanic_with_unix_datetime_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
41
42
43
44
45
46
47
import datetime
from typing import Callable, Generator, Union

from sanic import Sanic, json, response

from pait import field
from pait._pydanitc_adapter import is_v1
from pait.app.sanic import pait

if is_v1:

    class UnixDatetime(datetime.datetime):
        @classmethod
        def __get_validators__(cls) -> Generator[Callable, None, None]:
            yield cls.validate

        @classmethod
        def validate(cls, v: Union[int, str]) -> datetime.datetime:
            if isinstance(v, str):
                v = int(v)
            return datetime.datetime.fromtimestamp(v)

else:
    from pydantic import BeforeValidator
    from typing_extensions import Annotated

    def validate(v: Union[int, str]) -> datetime.datetime:
        if isinstance(v, str):
            v = int(v)
        return datetime.datetime.fromtimestamp(v)

    UnixDatetime = Annotated[datetime.datetime, BeforeValidator(validate)]  # type: ignore


@pait()
async def demo(timestamp: UnixDatetime = field.Query.i()) -> response.HTTPResponse:
    return json({"time": timestamp.isoformat()})


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)
docs_source_code/docs_source_code/introduction/how_to_use_type/tornado_with_unix_datetime_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
41
42
43
44
45
46
47
48
49
50
51
import datetime
from typing import Callable, Generator, Union

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

from pait import field
from pait._pydanitc_adapter import is_v1
from pait.app.tornado import pait

if is_v1:

    class UnixDatetime(datetime.datetime):
        @classmethod
        def __get_validators__(cls) -> Generator[Callable, None, None]:
            yield cls.validate

        @classmethod
        def validate(cls, v: Union[int, str]) -> datetime.datetime:
            if isinstance(v, str):
                v = int(v)
            return datetime.datetime.fromtimestamp(v)

else:
    from pydantic import BeforeValidator
    from typing_extensions import Annotated

    def validate(v: Union[int, str]) -> datetime.datetime:
        if isinstance(v, str):
            v = int(v)
        return datetime.datetime.fromtimestamp(v)

    UnixDatetime = Annotated[datetime.datetime, BeforeValidator(validate)]  # type: ignore


class DemoHandler(RequestHandler):
    @pait()
    def get(self, timestamp: UnixDatetime = field.Query.i()) -> None:
        self.write({"time": timestamp.isoformat()})


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


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

重新运行修改后的代码并使用调用curl命令后可以发现返回的时间值已经没有带时区了：

➜  ~ curl "http://127.0.0.1:8000/api/demo?timestamp=1600000000"
{"time":"2020-09-13T20:26:40"}