跳转至

其它

1.隐式引入与显式引入

pait支持多个Web框架, 如果项目对应的依赖环境中只安装了其中的一个框架, 那么可以直接使用隐式引入:

from pait.app.any import pait, load_app, add_simple_route

但是如果同时安装了多个框架, 那么隐式import将会引发异常, 建议使用显示引入，如下:

from pait.app.flask import pait, load_app, add_simple_route
from pait.app.sanic import pait, load_app, add_simple_route
from pait.app.starlette import pait, load_app, add_simple_route
from pait.app.tornado import pait, load_app, add_simple_route

2.内部方法

Pait内部封装了一些通用的方法，通过这些方法，开发者可以在不考虑兼容不同的Web框架的情况下快速的开发出拓展包，或者对Pait进行拓展， OpenAPI路由和grpc-gateway就是基于这些方法开发的。

2.1.data

data是每个CoreModel的载体，Pait装饰路由函数时会生成一个CoreModel并存放在pait.g.data中，以便为配置，文档生成等功能提供支持。

2.2.load_app

CoreModel会存储很多路由函数的信息, 但是路由函数缺少关键的OpenAPI信息数据如url, method等， 所以在使用OpenAPI之前还需要使用load_app补全数据，它的使用方法很简单，不过需要要在注册所有路由后再调用，如下:

Note

OpenAPI 路由在初始化之前会自动调用load_app

from flask import Flask

from pait.app.flask import load_app

app: Flask = Flask()

load_app(app) # 错误的使用方法
# --------
# app.add_url_rule
# --------

load_app(app) # 正确的使用方法
app.run()

import uvicorn
from starlette.applications import Starlette

from pait.app.starlette import load_app

app: Starlette = Starlette()
# 错误的使用方法
load_app(app)
# --------
# app.add_route
# --------

# 正确的使用方法
load_app(app)
uvicorn.run(app)

from sanic import Sanic

from pait.app.sanic import load_app

app: Sanic = Sanic()

load_app(app) # 错误的使用方法

# --------
# app.add_route
# --------

load_app(app) # 正确的使用方法
app.run()

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

from pait.app.tornado import load_app

app: Application = Application()

load_app(app) # 错误的使用方法
# --------
# app.add_handlers
# --------
load_app(app) # 正确的使用方法
app.listen(8000)
IOLoop.instance().start()

2.3.HTTP异常

Pait为每个Web框架提供了一个HTTP异常生成函数，它通过HTTP状态码，错误内容，Headers等参数生成Web框架的HTTP标准异常，它们的使用方法如下:

from pait.app.flask import http_exception

http_exception(status_code=401, message="Unauthorized", headers={"WWW-Authenticate": "Basic"})

from pait.app.sanic import http_exception

http_exception(status_code=401, message="Unauthorized", headers={"WWW-Authenticate": "Basic"})

from pait.app.starlette import http_exception

http_exception(status_code=401, message="Unauthorized", headers={"WWW-Authenticate": "Basic"})

from pait.app.tornado import http_exception

http_exception(status_code=401, message="Unauthorized", headers={"WWW-Authenticate": "Basic"})

此外，Pait还提供了一套常见的HTTP异常响应的Model，如下:

from pait.app.any import pait
from pait.model import response

# response.Http400RespModel
# response.Http401RespModel
# response.Http403RespModel
# response.Http404RespModel
# response.Http405RespModel
# response.Http406RespModel
# response.Http407RespModel
# response.Http408RespModel
# response.Http429RespModel

@pait(response_model_list=[response.Http400RespModel])
def demo() -> None:
    pass

同时HTTP异常响应的Model也支持自定义创建，如下使用示例:

from pait.model import response

# 创建一个状态码为500,content-type为html的响应Model
response.HttpStatusCodeBaseModel.clone(resp_model=response.HtmlResponseModel, status_code=500)
# 创建一个状态码为500,content-type为text的响应Model
response.HttpStatusCodeBaseModel.clone(resp_model=response.TextResponseModel, status_code=500)

2.4.SimpleRoute

Pait通过SimpleRoute统一了不同Web框架的路由注册以及生成响应的方法。 开发者通过SimpleRoute可以在不考虑兼容的情况下很方便的完成路由创建和注册。

Note

统一的路由响应生成功能由UnifiedResponsePluginProtocol插件提供， 路由函数被注册时会为路由函数添加UnifiedResponsePluginProtocol插件

SimpleRoute的使用方法如下:

docs_source_code/other/flask_with_simple_route_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 flask import Flask

from pait.app.any import SimpleRoute, add_multi_simple_route, add_simple_route
from pait.app.flask import pait
from pait.model import response


@pait(response_model_list=[response.JsonResponseModel])
def json_route() -> dict:
    return {}


@pait(response_model_list=[response.TextResponseModel])
def text_route() -> str:
    return "demo"


@pait(response_model_list=[response.HtmlResponseModel])
def html_route() -> str:
    return "<h1>demo</h1>"


app: Flask = Flask("demo")
add_simple_route(app, SimpleRoute(route=json_route, url="/json", methods=["GET"]))
add_multi_simple_route(
    app,
    SimpleRoute(route=json_route, url="/json", methods=["GET"]),
    SimpleRoute(route=text_route, url="/text", methods=["GET"]),
    SimpleRoute(route=html_route, url="/html", methods=["GET"]),
    prefix="/api",
    title="api",
)


if __name__ == "__main__":
    app.run(port=8000)
docs_source_code/other/starlette_with_simple_route_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 starlette.applications import Starlette

from pait.app.any import SimpleRoute, add_multi_simple_route, add_simple_route
from pait.app.starlette import pait
from pait.model import response


@pait(response_model_list=[response.JsonResponseModel])
async def json_route() -> dict:
    return {}


@pait(response_model_list=[response.TextResponseModel])
async def text_route() -> str:
    return "demo"


@pait(response_model_list=[response.HtmlResponseModel])
async def html_route() -> str:
    return "<h1>demo</h1>"


app: Starlette = Starlette()
add_simple_route(app, SimpleRoute(route=json_route, url="/json", methods=["GET"]))
add_multi_simple_route(
    app,
    SimpleRoute(route=json_route, url="/json", methods=["GET"]),
    SimpleRoute(route=text_route, url="/text", methods=["GET"]),
    SimpleRoute(route=html_route, url="/html", methods=["GET"]),
    prefix="/api",
    title="api",
)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)
docs_source_code/other/sanic_with_simple_route_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 sanic import Sanic
from sanic.request import Request

from pait.app.any import SimpleRoute, add_multi_simple_route, add_simple_route
from pait.app.sanic import pait
from pait.model import response


@pait(response_model_list=[response.JsonResponseModel])
async def json_route(request: Request) -> dict:
    return {}


@pait(response_model_list=[response.TextResponseModel])
async def text_route(request: Request) -> str:
    return "demo"


@pait(response_model_list=[response.HtmlResponseModel])
async def html_route(request: Request) -> str:
    return "<h1>demo</h1>"


app: Sanic = Sanic("demo")
add_simple_route(app, SimpleRoute(route=json_route, url="/json", methods=["GET"]))
add_multi_simple_route(
    app,
    SimpleRoute(route=json_route, url="/json", methods=["GET"]),
    SimpleRoute(route=text_route, url="/text", methods=["GET"]),
    SimpleRoute(route=html_route, url="/html", methods=["GET"]),
    prefix="/api",
    title="api",
)


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)
docs_source_code/other/tornado_with_simple_route_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 tornado.ioloop import IOLoop
from tornado.web import Application

from pait.app.any import SimpleRoute, add_multi_simple_route, add_simple_route
from pait.app.tornado import pait
from pait.model import response


@pait(response_model_list=[response.JsonResponseModel])
async def json_route() -> dict:
    return {}


@pait(response_model_list=[response.TextResponseModel])
async def text_route() -> str:
    return "demo"


@pait(response_model_list=[response.HtmlResponseModel])
async def html_route() -> str:
    return "<h1>demo</h1>"


app: Application = Application()
add_simple_route(app, SimpleRoute(route=json_route, url="/json", methods=["GET"]))
add_multi_simple_route(
    app,
    SimpleRoute(route=json_route, url="/json", methods=["GET"]),
    SimpleRoute(route=text_route, url="/text", methods=["GET"]),
    SimpleRoute(route=html_route, url="/html", methods=["GET"]),
    prefix="/api",
    title="api",
)


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

第一段高亮代码是按照SimpleRoute标准创建了三个路由函数，SimpleRoute的标准如下:

  • 1.路由函数需要被pait装饰，同时response_model_list属性不能为空（代码中路由函数的响应模型分别为JsonResponseModel，TextResponseModel，HtmlResponseModel，这些都是SimpleRoute要求的，如果没有响应模型，那么SimpleRoute无法把路由函数注册到Web框架中。）
  • 2.路由函数的返回值从响应对象变为是Python的基础类型，返回的Python基础类型需要跟响应模型的response_data保持一致。

第二段高亮是通过add_simple_route和add_multi_simple_route方法注册路由，其中add_simple_route只能注册一个路由，而add_multi_simple_route可以注册多个路由，它们的都接收app和SimpleRoute实例，而SimpleRoute只支持三个属性，如下:

参数 描述
route 符合SimpleRoute标准的路由函数
url 当前路由的Url
method 当前路由对应的HTTP Method

此外，add_multi_simple_route还支持两个可选参数，如下:

参数 描述
prefix 路由前缀，比如prefix为"/api"，SimpleRoute的url为"/user"时，注册的路由URL为"/api/user"
title 当前路由组的标题，对于某些框架，它们采用的路由组或者蓝图都需要有唯一的命名，所以不同add_multi_simple_route的title都应该不同

在运行代码后，通过curl命令测试路由可以正常工作:

curl http://127.0.0.1:8000/json{} curl http://127.0.0.1:8000/api/json{} curl http://127.0.0.1:8000/api/textdemo curl http://127.0.0.1:8000/api/html<h1>demo</h1>

2.5.设置与获取Web框架属性

Pait为Web框架的设置与获取Web框架属性值的方法提供了一个统一的方法，它们分别是set_app_attribute和get_app_attribute， 通过set_app_attribute和get_app_attribute可以在任一时刻设置与获取Web框架属性，使用方法如下：

docs_source_code/other/flask_with_attribute_demo.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
import httpx
from flask import Flask, Response, current_app, jsonify

from pait.app.any import get_app_attribute, set_app_attribute


def demo_route() -> Response:
    client: httpx.Client = get_app_attribute(current_app, "client")
    return jsonify({"status_code": client.get("http://so1n.me").status_code})


app: Flask = Flask("demo")
app.add_url_rule("/api/demo", "demo", demo_route, methods=["GET"])
set_app_attribute(app, "client", httpx.Client())


if __name__ == "__main__":
    app.run(port=8000)
docs_source_code/other/starlette_with_attribute_demo.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
import httpx
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse

from pait.app.any import get_app_attribute, set_app_attribute


async def demo_route(request: Request) -> JSONResponse:
    client: httpx.AsyncClient = get_app_attribute(request.app, "client")
    return JSONResponse({"status_code": (await client.get("http://so1n.me")).status_code})


app: Starlette = Starlette()
app.add_route("/api/demo", demo_route, methods=["GET"])
set_app_attribute(app, "client", httpx.AsyncClient())


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)
docs_source_code/other/sanic_with_attribute_demo.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
import httpx
from sanic import HTTPResponse, Request, Sanic, json

from pait.app.any import get_app_attribute, set_app_attribute


async def demo_route(request: Request) -> HTTPResponse:
    client: httpx.AsyncClient = get_app_attribute(request.app, "client")
    return json({"status_code": (await client.get("http://so1n.me")).status_code})


app: Sanic = Sanic("demo")
app.add_route(demo_route, "/api/demo", methods=["GET"])
set_app_attribute(app, "client", httpx.AsyncClient())


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)
docs_source_code/other/tornado_with_attribute_demo.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
import httpx
from tornado.ioloop import IOLoop
from tornado.web import Application, RequestHandler

from pait.app.any import get_app_attribute, set_app_attribute


class DemoHandler(RequestHandler):
    async def get(self) -> None:
        client: httpx.AsyncClient = get_app_attribute(self.application, "client")
        self.write({"status_code": (await client.get("http://so1n.me")).status_code})


app: Application = Application()
app.add_handlers(".*$", [("/api/demo", DemoHandler)])
set_app_attribute(app, "client", httpx.AsyncClient())


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

在运行代码后，可以通过以下命令进行测试:

➜  curl http://127.0.0.1:8000/api/demo
{"status_code": 200}

通过结果可以看到，路由函数能够获取到client并通过client获取到url的status_code。

Note

通过为Web框架设置属性值，可以使组件与框架解耦，同时也可以使组件更加灵活，但是更加推荐通过DI工具来实现解耦，具体的DI工具见Awesome Dependency Injection in Python。

3.如何在其它Web框架使用Pait

目前Pait还在快速迭代中，所以还是以功能开发为主，如果要在其他尚未支持的框架中使用Pait, 或者要对功能进行拓展, 可以参照两个框架进行简单的适配即可.

同步类型的web框架请参照 pait.app.flask

异步类型的web框架请参照 pait.app.starlette

4.示例代码

更多完整示例请参考example

5.发行说明

详细的发版说明见CHANGELOG