跳转至

介绍

Pait属于辅助型框架，它并不会对Web框架的原有使用方式进行明显的改变，所以在介绍Pait的使用之前，先看看不同Web框架的使用方式。

docs_source_code/docs_source_code/introduction/flask_hello_world_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 flask import Flask, Response, jsonify, request


def demo_post() -> Response:
    request_dict = request.json or {}
    uid_str: str = request_dict.get("uid", "")
    username: str = request_dict.get("username", "")

    uid = int(uid_str)  # Don't think about type conversion exceptions for now
    if not (10 < uid < 1000):
        raise ValueError("invalid uid")
    if not (2 <= len(username) <= 4):
        raise ValueError("invalid name")
    # get the corresponding value and return it
    return jsonify({"uid": uid, "username": username})


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


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


async def demo_post(request: Request) -> JSONResponse:
    request_dict = await request.json()
    uid_str: str = request_dict.get("uid", "")
    username: str = request_dict.get("username", "")

    uid = int(uid_str)  # Don't think about type conversion exceptions for now
    if not (10 < uid < 1000):
        raise ValueError("invalid uid")
    if not (2 <= len(username) <= 4):
        raise ValueError("invalid name")
    # get the corresponding value and return it
    return JSONResponse({"uid": uid, "username": username})


app = Starlette(routes=[Route("/api", demo_post, methods=["POST"])])


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)
docs_source_code/docs_source_code/introduction/sanic_hello_world_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
from sanic.app import Sanic
from sanic.request import Request
from sanic.response import HTTPResponse, json


async def demo_post(request: Request) -> HTTPResponse:
    request_dict = await request.json()
    uid_str: str = request_dict.get("uid", "")
    username: str = request_dict.get("username", "")

    uid = int(uid_str)  # Don't think about type conversion exceptions for now
    if not (10 < uid < 1000):
        raise ValueError("invalid uid")
    if not (2 <= len(username) <= 4):
        raise ValueError("invalid name")
    # get the corresponding value and return it
    return json({"uid": uid, "username": username})


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


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)
docs_source_code/docs_source_code/introduction/tornado_hello_world_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
import json

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


class DemoHandler(RequestHandler):
    def post(self) -> None:
        request_dict = json.loads(self.request.body.decode())
        uid_str: str = request_dict.get("uid", "")
        username: str = request_dict.get("username", "")

        uid = int(uid_str)  # Don't think about type conversion exceptions for now
        if not (10 < uid < 1000):
            raise ValueError("invalid uid")
        if not (2 <= len(username) <= 4):
            raise ValueError("invalid name")
        # get the corresponding value and return it
        self.write({"uid": uid, "username": username})


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


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

这段示例代码的与首页的示例代码的逻辑是一致的，示例代码的主要功能是在启动时将一个路由注册到Web框架的实例中，并在运行时收到一个url为/api，method为POST的请求后会把请求交由路由函数处理。 而路由函数的处理逻辑也很简单，它会先进行数据校验，当数据在符合要求的情况下才会返回，否则会直接抛出错误。

接下来，将在示例代码中使用Pait，最终它们的功能是一样的，代码如下：

docs_source_code/docs_source_code/introduction/flask_demo.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
from typing import Type

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

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


@pait()
def demo_post(
    uid: int = Json.t(description="user id", gt=10, lt=1000),
    username: str = Json.t(description="user name", min_length=2, max_length=4),
) -> Response:
    return jsonify({"uid": uid, "user_name": username})


app = Flask("demo")
app.add_url_rule("/api", "demo", demo_post, methods=["POST"])
app.run(port=8000)
docs_source_code/docs_source_code/introduction/starlette_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 typing import Type

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

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


@pait()
async def demo_post(
    uid: int = Json.t(description="user id", gt=10, lt=1000),
    username: str = Json.t(description="user name", min_length=2, max_length=4),
) -> JSONResponse:
    return JSONResponse({"uid": uid, "user_name": username})


app = Starlette(routes=[Route("/api", demo_post, methods=["POST"])])

import uvicorn
uvicorn.run(app)
docs_source_code/docs_source_code/introduction/sanic_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 typing import Type

from pydantic import BaseModel, Field
from sanic.app import Sanic
from sanic.response import HTTPResponse, json

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


@pait()
async def demo_post(
    uid: int = Json.t(description="user id", gt=10, lt=1000),
    username: str = Json.t(description="user name", min_length=2, max_length=4),
) -> HTTPResponse:
    return json({"uid": uid, "user_name": username})


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

import uvicorn
uvicorn.run(app)
docs_source_code/docs_source_code/introduction/tornado_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 Type

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

from pait.app.tornado import pait
from pait.field import Json
from pait.model.response import JsonResponseModel
from pait.openapi.doc_route import AddDocRoute


class DemoResponseModel(JsonResponseModel):
    class ResponseModel(BaseModel):
        uid: int = Field()
        user_name: str = Field()

    description: str = "demo response"
    response_data: Type[BaseModel] = ResponseModel


class DemoHandler(RequestHandler):
    @pait(response_model_list=[DemoResponseModel])
    def post(
        self,
        uid: int = Json.t(description="user id", gt=10, lt=1000),
        username: str = Json.t(description="user name", min_length=2, max_length=4),
    ) -> None:
        self.write({"uid": uid, "user_name": username})


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


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

示例代码中第一段高亮代码的@pait装饰器是Pait所有功能的核心，在使用@pait装饰路由函数后，Pait会通过inspect获取到函数签名并生成依赖注入规则。 比如第二段高亮代码中路由函数的参数都以<name>:<type>=<default>格式的关键参数填写，Pait在初始化的时候自动通过如下的规则将关键参数转化为自己的依赖注入规则：

key 含义 作用
name 参数名 Pait会以name为Key从请求资源获取对应的值
type 参数类型 用于参数校验或者转化的类型
default Pait的Field对象 不同的Field类型代表从不同的请求类型获取值;Field对象的属性则告诉Pait该如何从请求中获取的值，并进行校验。

以上面的uid参数为例子，Pait会通过Json从请求中获取Json数据，接着以uid为Key从Json数据中获取对应的的值并转化并验证是否为int类型， 最后再判断该值是否处于10-1000之间，如果不是就直接报错， 如果是则赋值给uid变量。

通过Hello World代码与使用Pait后的代码做对比，可以看到使用Pait后的代码更加简单明了，同时也符合了现在Python流行的Type Hint，提高了代码的稳健性。

Note

直接使用Json()时, mypy会检查到类型不匹配, 为此可以通过Json.i()忽略这个问题。 如果需要mypy检查Json中default，default_factory以及example属性的值， 那么建议直接使用Json.t()。