Depend

前文提到的Field对象都是与请求对象相关的，他们的作用都是把请求对象指定的资源注入到路由函数中。而Depend则是一种特殊Field对象，他可以把符合Pait规则的函数注入到路由函数中，它可以实现如下功能:

共享相同的逻辑
实现安全校验的功能
与别的系统交互(如数据库)。

Note

Depend只做请求对象相关的依赖注入，无法完成请求对象之外的依赖注入功能。如果你有这方面的需求，推荐通过DI工具来实现依赖注入功能，具体的DI工具见Awesome Dependency Injection in Python。

1.Depend的使用

通常情况下，业务系统都会有用户Token校验的功能，这个功能是非常符合Depend的使用场景。在这个场景中，用户每次访问系统时都需要带上Token，而服务端在收到用户的请求后会先判断Token是否合法，合法则会放行，不合法则会返回错误信息。

大多数类Flask微Web框架使用者都会选择使用Python装饰器来解决这个问题，如下:

@check_token()
def demo_route() -> None:
    pass

有些时候还会增加一些功能，比如根据Token去获取到uid数据并传给路由函数:

@check_token()
def demo_route(uid: str) -> None:
    pass

不过可以看出这种实现方法比较动态，它会导致代码检测工具很难检测这段代码是否存在问题。只有在拥有良好的内部规范才有可能防止开发人员错误的使用check_token装饰器，但它也没办法完全防止check_token装饰器被错误的使用。

使用Pait的Depend可以解决这个问题，Pait的Depend使用示例代码如下:

FlaskStarletteSanicTornado

docs_source_code/introduction/depend/flask_with_depend_demo.py
from flask import Flask, Response, jsonify

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
    return jsonify({"data": str(exc)})


fake_db_dict: dict = {"u12345": "so1n"}


def get_user_by_token(token: str = field.Header.i()) -> str:
    if token not in fake_db_dict:
        raise RuntimeError(f"Can not found by token:{token}")
    return fake_db_dict[token]


@pait()
def demo(token: str = field.Depends.i(get_user_by_token)) -> dict:
    return {"user": token}


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/introduction/depend/starlette_with_depend_demo.py
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
    return JSONResponse({"data": str(exc)})


fake_db_dict: dict = {"u12345": "so1n"}


async def get_user_by_token(token: str = field.Header.i()) -> str:
    if token not in fake_db_dict:
        raise RuntimeError(f"Can not found by token:{token}")
    return fake_db_dict[token]


@pait()
async def demo(token: str = field.Depends.i(get_user_by_token)) -> JSONResponse:
    return JSONResponse({"user": token})


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/introduction/depend/sanic_with_depend_demo.py
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
    return json({"data": str(exc)})


fake_db_dict: dict = {"u12345": "so1n"}


async def get_user_by_token(token: str = field.Header.i()) -> str:
    if token not in fake_db_dict:
        raise RuntimeError(f"Can not found by token:{token}")
    return fake_db_dict[token]


@pait()
async def demo(token: str = field.Depends.i(get_user_by_token)) -> HTTPResponse:
    return json({"user": token})


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/introduction/depend/tornado_with_depend_demo.py
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({"data": str(exc)})
        self.finish()


fake_db_dict: dict = {"u12345": "so1n"}


async def get_user_by_token(token: str = field.Header.i()) -> str:
    if token not in fake_db_dict:
        raise RuntimeError(f"Can not found by token:{token}")
    return fake_db_dict[token]


class DemoHandler(_Handler):
    @pait()
    async def get(self, token: str = field.Depends.i(get_user_by_token)) -> None:
        self.write({"user": token})


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


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

示例代码中第一段高亮代码是模仿数据库的调用方法，目前假设数据库只有用户so1n拥有token，且token值为"u12345"。第二段高亮代码是一个名为get_user_by_token的函数，它负责从Header中获取Token并校验Token是否存在，如果存在则返回用户,不存在则抛错。这个函数是一个特殊的函数，它的参数填写规则与被Pait装饰的路由函数一致，所以之前提到的任何写法都可以在这个函数中使用，同时该函数可以被Pait的Depend使用。第三段高亮代码则是路由函数填写的Token参数，这里比较特殊的是通过field.Depend来裹住get_user_by_token函数，这样Pait就能够知道当前路由函数的Token参数必须通过get_user_by_token函数获取。

在运行代码并调用curl命令可以发现发现这段代码工作一切正常，当token存在时返回用户，不存在则返回抛错信息: After running the code and calling the curl command, can find that this code works normally. When the token exists, it returns to the user. If it does not exist, it returns an error message:

curl "http://127.0.0.1:8000/api/demo" --header "token:u12345"{"user":"so1n"}curl "http://127.0.0.1:8000/api/demo" --header "token:u123456"{"data":"Can not found by token:u123456"}

除此之外，Pait还能支持多层Depend嵌套的。以上面的代码为例子，现在假设需要先校验Token合法后才会去数据库获取对应的用户，代码可以进行如下改写：

FlaskStarletteSanicTornado

docs_source_code/introduction/depend/flask_with_nested_depend_demo.py
from flask import Flask, Response, jsonify

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
    return jsonify({"data": str(exc)})


fake_db_dict: dict = {"u12345": "so1n"}


def check_token(token: str = field.Header.i()) -> str:
    if len(token) != 6 and token[0] != "u":
        raise RuntimeError("Illegal Token")
    return token


def get_user_by_token(token: str = field.Depends.i(check_token)) -> str:
    if token not in fake_db_dict:
        raise RuntimeError(f"Can not found by token:{token}")
    return fake_db_dict[token]


@pait()
def demo(token: str = field.Depends.i(get_user_by_token)) -> dict:
    return {"user": token}


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/introduction/depend/starlette_with_nested_depend_demo.py
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
    return JSONResponse({"data": str(exc)})


fake_db_dict: dict = {"u12345": "so1n"}


def check_token(token: str = field.Header.i()) -> str:
    if len(token) != 6 and token[0] != "u":
        raise RuntimeError("Illegal Token")
    return token


async def get_user_by_token(token: str = field.Depends.i(check_token)) -> str:
    if token not in fake_db_dict:
        raise RuntimeError(f"Can not found by token:{token}")
    return fake_db_dict[token]


@pait()
async def demo(token: str = field.Depends.i(get_user_by_token)) -> JSONResponse:
    return JSONResponse({"user": token})


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/introduction/depend/sanic_with_nested_depend_demo.py
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
    return json({"data": str(exc)})


fake_db_dict: dict = {"u12345": "so1n"}


def check_token(token: str = field.Header.i()) -> str:
    if len(token) != 6 and token[0] != "u":
        raise RuntimeError("Illegal Token")
    return token


async def get_user_by_token(token: str = field.Depends.i(check_token)) -> str:
    if token not in fake_db_dict:
        raise RuntimeError(f"Can not found by token:{token}")
    return fake_db_dict[token]


@pait()
async def demo(token: str = field.Depends.i(get_user_by_token)) -> HTTPResponse:
    return json({"user": token})


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/introduction/depend/tornado_with_nested_depend_demo.py
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({"data": str(exc)})
        self.finish()


fake_db_dict: dict = {"u12345": "so1n"}


def check_token(token: str = field.Header.i()) -> str:
    if len(token) != 6 and token[0] != "u":
        raise RuntimeError("Illegal Token")
    return token


async def get_user_by_token(token: str = field.Depends.i(check_token)) -> str:
    if token not in fake_db_dict:
        raise RuntimeError(f"Can not found by token:{token}")
    return fake_db_dict[token]


class DemoHandler(_Handler):
    @pait()
    async def get(self, token: str = field.Depends.i(get_user_by_token)) -> None:
        self.write({"user": token})


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


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

示例代码中的高亮代码为本次修改后的代码，这部分代码主要是新增了一个check_token的函数用来获取和校验Token，同时get_user_by_token获取Token的来源从Header变为check_token。

在运行代码并调用curl命令进行测试，通过输出结果可以发现不符合校验逻辑的会返回抛错信息：

curl "http://127.0.0.1:8000/api/demo" --header "token:u12345"{"user":"so1n"}curl "http://127.0.0.1:8000/api/demo" --header "token:u123456"{"data":"Can not found by token:u123456"}curl "http://127.0.0.1:8000/api/demo" --header "token:fu12345"{"data":"Illegal Token"}

2.基于ContextManager的Depend

上文所示的Depends用法虽然都能够正常的运行，但是它们没办法像Python装饰器一样知道函数的运行情况，包括函数是否正常运行，产生的异常是什么，何时运行结束等等等，这时就需要基于ContextManager的Depend来解决这个问题。

基于ContextManager的Depend使用很简单，只要把函数加上对应的ContextManager装饰器，然后按照ContextManager官方文档中描述使用try,except,finally语法块即可，如下示例代码:

from contextlib import contextmanager
from typing import Any, Generator

@contextmanager
def demo() -> Generator[Any, Any, Any]:
    try:
        # 1
        yield None
    except Exception:
        # 2
        pass
    finally:
        # 3
        pass

该示例代码中序号1的位置用来编写正常的函数逻辑，并通过yield返回数据。序号2的位置用来编写当函数运行异常时的代码逻辑。最后的序号3则是用来编写统一的函数运行结束处理逻辑。

下面的代码是一个使用了ContextManager和Depend的例子：

FlaskStarletteSanicTornado

docs_source_code/introduction/depend/flask_with_context_manager_depend_demo.py
from contextlib import contextmanager
from typing import Any, Generator

from flask import Flask, Response, jsonify

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
    return jsonify({"data": str(exc)})


class _DemoSession(object):
    def __init__(self, uid: int) -> None:
        self._uid: int = uid
        self._status: bool = False

    @property
    def uid(self) -> int:
        if self._status:
            return self._uid
        else:
            raise RuntimeError("Session is close")

    def create(self) -> None:
        self._status = True

    def close(self) -> None:
        self._status = False


@contextmanager
def context_depend(uid: int = field.Query.i(description="user id", gt=10, lt=1000)) -> Generator[int, Any, Any]:
    session: _DemoSession = _DemoSession(uid)
    try:
        print("context_depend init")
        session.create()
        yield session.uid
    except Exception:
        print("context_depend error")
    finally:
        print("context_depend exit")
        session.close()


@pait()
def demo(uid: int = field.Depends.i(context_depend), is_raise: bool = field.Query.i(default=False)) -> Response:
    if is_raise:
        raise RuntimeError()
    return jsonify({"code": 0, "msg": uid})


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/introduction/depend/starlette_with_context_manager_depend_demo.py
from contextlib import asynccontextmanager
from typing import Any, AsyncGenerator

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
    return JSONResponse({"data": str(exc)})


class _DemoSession(object):
    def __init__(self, uid: int) -> None:
        self._uid: int = uid
        self._status: bool = False

    @property
    def uid(self) -> int:
        if self._status:
            return self._uid
        else:
            raise RuntimeError("Session is close")

    def create(self) -> None:
        self._status = True

    def close(self) -> None:
        self._status = False


@asynccontextmanager
async def context_depend(uid: int = field.Query.i(description="user id", gt=10, lt=1000)) -> AsyncGenerator[int, Any]:
    session: _DemoSession = _DemoSession(uid)
    try:
        print("context_depend init")
        session.create()
        yield session.uid
    except Exception:
        print("context_depend error")
    finally:
        print("context_depend exit")
        session.close()


@pait()
async def demo(
    uid: int = field.Depends.i(context_depend), is_raise: bool = field.Query.i(default=False)
) -> JSONResponse:
    if is_raise:
        raise RuntimeError()
    return JSONResponse({"code": 0, "msg": uid})


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/introduction/depend/sanic_with_context_manager_depend_demo.py
from contextlib import asynccontextmanager
from typing import Any, AsyncGenerator

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
    return json({"data": str(exc)})


class _DemoSession(object):
    def __init__(self, uid: int) -> None:
        self._uid: int = uid
        self._status: bool = False

    @property
    def uid(self) -> int:
        if self._status:
            return self._uid
        else:
            raise RuntimeError("Session is close")

    def create(self) -> None:
        self._status = True

    def close(self) -> None:
        self._status = False


@asynccontextmanager
async def context_depend(uid: int = field.Query.i(description="user id", gt=10, lt=1000)) -> AsyncGenerator[int, Any]:
    session: _DemoSession = _DemoSession(uid)
    try:
        print("context_depend init")
        session.create()
        yield session.uid
    except Exception:
        print("context_depend error")
    finally:
        print("context_depend exit")
        session.close()


@pait()
async def demo(
    uid: int = field.Depends.i(context_depend), is_raise: bool = field.Query.i(default=False)
) -> HTTPResponse:
    if is_raise:
        raise RuntimeError()
    return json({"code": 0, "msg": uid})


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/introduction/depend/tornado_with_context_manager_depend_demo.py
from contextlib import asynccontextmanager
from typing import Any, AsyncGenerator

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 _DemoSession(object):
    def __init__(self, uid: int) -> None:
        self._uid: int = uid
        self._status: bool = False

    @property
    def uid(self) -> int:
        if self._status:
            return self._uid
        else:
            raise RuntimeError("Session is close")

    def create(self) -> None:
        self._status = True

    def close(self) -> None:
        self._status = False


@asynccontextmanager
async def context_depend(uid: int = field.Query.i(description="user id", gt=10, lt=1000)) -> AsyncGenerator[int, Any]:
    session: _DemoSession = _DemoSession(uid)
    try:
        print("context_depend init")
        session.create()
        yield session.uid
    except Exception:
        print("context_depend error")
    finally:
        print("context_depend exit")
        session.close()


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()
    async def get(
        self, uid: int = field.Depends.i(context_depend), is_raise: bool = field.Query.i(default=False)
    ) -> None:
        if is_raise:
            raise RuntimeError()
        self.write({"code": 0, "msg": uid})


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


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

该示例假设每次调用请求时都会基于对应的uid创建一个Session且Session会在请求结束时自动关闭. 其中第一段高亮代码是模拟一个基于uid的Session; 第二段高亮代码则是一个被ContextManger装饰的Depends函数，它会在try, except以及finally打印不同的内容; 而第三段高亮代码则是路由函数，它会依据参数is_raise是否为True来决定抛错还是正常返回。

现在运行代码并使用curl进行接口测试，发现第一个请求的响应结果是正常的，而第二个请求发生异常(返回空字符串)：

curl "http://127.0.0.1:8000/api/demo?uid=999"{"code":0,"msg":999}curl "http://127.0.0.1:8000/api/demo?uid=999&is_raise=True"{"data":""}

这时切回到运行示例代码的终端，可以发现终端打印了类似如下数据:

context_depend init
context_depend exit
INFO:     127.0.0.1:44162 - "GET /api/demo?uid=999 HTTP/1.1" 200 OK
context_depend init
context_depend error
context_depend exit
INFO:     127.0.0.1:44164 - "GET /api/demo?uid=999&is_raise=True HTTP/1.1" 200 OK

通过终端输出的数据可以看出，在第一次请求时, 终端只打印了init和exit，而在第二次请求时，终端会在init和exit中间多打印了一行error。

3.基于类的Depend

基于类的Depend与基于函数的Depend类似，它们之间的区别是Pait不但会解析类的__call__方法的函数签名之外，还会去解析类的属性，如下示例：

FlaskStarletteSanicTornado

docs_source_code/introduction/depend/flask_with_class_depend_demo.py
from flask import Flask, Response, jsonify

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
    return jsonify({"data": str(exc)})


fake_db_dict: dict = {"u12345": "so1n"}


class GetUserDepend(object):
    user_name: str = field.Query.i()

    def __call__(self, token: str = field.Header.i()) -> str:
        if token not in fake_db_dict:
            raise RuntimeError(f"Can not found by token:{token}")
        user_name = fake_db_dict[token]
        if user_name != self.user_name:
            raise RuntimeError("The specified user could not be found through the token")
        return user_name


@pait()
def demo(token: str = field.Depends.i(GetUserDepend)) -> dict:
    return {"user": token}


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/introduction/depend/starlette_with_class_depend_demo.py
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
    return JSONResponse({"data": str(exc)})


fake_db_dict: dict = {"u12345": "so1n"}


class GetUserDepend(object):
    user_name: str = field.Query.i()

    async def __call__(self, token: str = field.Header.i()) -> str:
        if token not in fake_db_dict:
            raise RuntimeError(f"Can not found by token:{token}")
        user_name = fake_db_dict[token]
        if user_name != self.user_name:
            raise RuntimeError("The specified user could not be found through the token")
        return user_name


@pait()
async def demo(token: str = field.Depends.i(GetUserDepend)) -> JSONResponse:
    return JSONResponse({"user": token})


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/introduction/depend/sanic_with_class_depend_demo.py
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
    return json({"data": str(exc)})


fake_db_dict: dict = {"u12345": "so1n"}


class GetUserDepend(object):
    user_name: str = field.Query.i()

    async def __call__(self, token: str = field.Header.i()) -> str:
        if token not in fake_db_dict:
            raise RuntimeError(f"Can not found by token:{token}")
        user_name = fake_db_dict[token]
        if user_name != self.user_name:
            raise RuntimeError("The specified user could not be found through the token")
        return user_name


@pait()
async def demo(token: str = field.Depends.i(GetUserDepend)) -> HTTPResponse:
    return json({"user": token})


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/introduction/depend/tornado_with_class_depend_demo.py
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({"data": str(exc)})
        self.finish()


fake_db_dict: dict = {"u12345": "so1n"}


class GetUserDepend(object):
    user_name: str = field.Query.i()

    async def __call__(self, token: str = field.Header.i()) -> str:
        if token not in fake_db_dict:
            raise RuntimeError(f"Can not found by token:{token}")
        user_name = fake_db_dict[token]
        if user_name != self.user_name:
            raise RuntimeError("The specified user could not be found through the token")
        return user_name


class DemoHandler(_Handler):
    @pait()
    async def get(self, token: str = field.Depends.i(GetUserDepend)) -> None:
        self.write({"user": token})


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


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

示例代码中的第一段高亮代码是基于类的Depend实现，这段代码主要分为两部分，第一部分是类的属性，这里也采用<name>: <type> = <default>的格式编写的，每当请求命中路由时，Pait都会为该类注入对应的值。第二部分代码是根据Depend的使用中的例子进行改写的，它会校验Token以及对应的用户名(正常的逻辑基本不会这样做，这里只做功能演示)，__call__方法的使用方法与基于函数的Depend类似。

__call__方法使用限制说明

在Python中万物皆对象，所以一个拥有__call__方法的类与函数类似，如下示例代码：

from typing import Any

class DemoDepend(object):
    def __call__(self, *args: Any, **kwargs: Any) -> Any:
        pass

代码中的__call__方法是一个直观的使用方式，但是由于Python的限制，__call__方法不支持函数签名重写，比如下面的例子:

from typing import Any
from pait import field

class DemoDepend(object):
    def __init__(self) -> Any:
        def new_call(uid: str = field.Query.i(), user_name: str = field.Query.i()) -> Any:
            pass
        setattr(self, "__call__", new_call)

    def __call__(self, uid: str = field.Query.i()) -> Any:
        pass

该类实例化后，inspect解析出来的__call__方法的函数签名仍然是__call__(self, uid: str = field.Query.i()) -> Any，而不是__call__(uid: str = field.Query.i(), user_name: str = field.Query.i()) -> Any。这会导致Pait无法提取正确的参数规则，为了解决这个问题，Pait会优先解析允许被重写的pait_handler方法，如下:

from typing import Any
from pait import field

class DemoDepend(object):
    def __init__(self) -> Any:
        def new_call(uid: str = field.Query.i(), user_name: str = field.Query.i()) -> Any:
            pass
        setattr(self, "pait_handler", new_call)

    def pait_handler(self, uid: str = field.Query.i()) -> Any:
        pass

该类实例化后，Pait就能正常解析出pait_handler的函数签名是pait_handler(uid: str = field.Query.i(), user_name: str = field.Query.i()) -> Any

而第二段高亮代码中则把Depend参数中的基于函数的Depend替换为基于类的Depend。

在运行代码并执行如下curl命令，可以看到如下输出:

➜ ~ curl "http://127.0.0.1:8000/api/demo" --header "token:u12345"
{"data":"Can not found user_name value"}
➜ ~ curl "http://127.0.0.1:8000/api/demo?user_name=so1n" --header "token:u12345"
{"user":"so1n"}
➜ ~ curl "http://127.0.0.1:8000/api/demo?user_name=faker" --header "token:u12345"
{"data":"The specified user could not be found through the token"}

基于类的Depend的初始化说明

由于每次请求都会创建一个新的实例，这意味着无法跟平常一样自定义初始化参数。这时可以采用pait.util.partial_wrapper绑定初始化参数，如下例子:

from pait import field
from pait.util import partial_wrapper

class GetUserDepend(object):
    user_name: str = field.Query.i()
    age: int = field.Query.i()

    def __init__(self, age_limit: int = 18) -> None:
        self.age_limit: int = age_limit

    def __call__(self, token: str = field.Header.i()) -> str:
        if token not in fake_db_dict:
            raise RuntimeError(f"Can not found by token:{token}")
        user_name = fake_db_dict[token]
        if user_name != self.user_name:
            raise RuntimeError("The specified user could not be found through the token")
        if self.age < self.age_limit:
            raise ValueError("Minors cannot access")
        return user_name


@pait()
def demo(user_name: str = field.Depends.i(partial_wrapper(GetUserDepend, age_limit=16))):
    pass

@pait()
def demo1(user_name: str = field.Depends.i(GetUserDepend)):
    pass

这个例子中每个路由函数针对使用者的限制年龄有所不同，其中demo是限制年龄小于16的用户不可以访问，而demo1限制小于18岁的用户是不可以访问，所以他们的GetUserDepend的初始化参数是不同的。

为此，demo函数采用了pait.util.partial_wrapper把初始化参数跟GetUserDepend绑定。 pait.util.partial_wrapper的作用与官方的functools.partial类似，唯一不同的是它支持PEP612，可以获得代码提示以及使用检查工具进行代码检查。

4.Pre-Depend

在一些场景下路由函数只需要Depends函数执行校验逻辑，并不需要Depends函数的返回值，那么这时候可能会考虑用变量名_来进行替代，如下:

@pait()
def demo(_: str = field.Depends.i(get_user_by_token)) -> None:
    pass

不过Python并不支持一个函数内出现多个相同名字的变量，这意味着有多个类似的参数时，无法把他们的变量名都改为_。

为此，Pait通过可选参数pre_depend_list来解决这个问题。它的使用方法很简单，只需要把Depend函数从参数迁移到Pait的pre_depend_list可选参数即可， Depend代码的逻辑和功能均不会被受到影响，修改后代码会变为如下（高亮代码为修改部分）：

FlaskStarletteSanicTornado

docs_source_code/docs_source_code/introduction/depend/flask_with_pre_depend_demo.py
from flask import Flask, Response, jsonify

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
    return jsonify({"data": str(exc)})


fake_db_dict: dict = {"u12345": "so1n"}


def get_user_by_token(token: str = field.Header.i()) -> str:
    if token not in fake_db_dict:
        raise RuntimeError(f"Can not found by token:{token}")
    return fake_db_dict[token]


@pait(pre_depend_list=[get_user_by_token])
def demo() -> dict:
    return {"msg": "success"}


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/docs_source_code/introduction/depend/starlette_with_pre_depend_demo.py
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
    return JSONResponse({"data": str(exc)})


fake_db_dict: dict = {"u12345": "so1n"}


async def get_user_by_token(token: str = field.Header.i()) -> str:
    if token not in fake_db_dict:
        raise RuntimeError(f"Can not found by token:{token}")
    return fake_db_dict[token]


@pait(pre_depend_list=[get_user_by_token])
async def demo() -> JSONResponse:
    return JSONResponse({"msg": "success"})


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/docs_source_code/introduction/depend/sanic_with_pre_depend_demo.py
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
    return json({"data": str(exc)})


fake_db_dict: dict = {"u12345": "so1n"}


async def get_user_by_token(token: str = field.Header.i()) -> str:
    if token not in fake_db_dict:
        raise RuntimeError(f"Can not found by token:{token}")
    return fake_db_dict[token]


@pait(pre_depend_list=[get_user_by_token])
async def demo(request: Request) -> HTTPResponse:
    return json({"msg": "success"})


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/docs_source_code/introduction/depend/tornado_with_pre_depend_demo.py
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({"data": str(exc)})
        self.finish()


fake_db_dict: dict = {"u12345": "so1n"}


async def get_user_by_token(token: str = field.Header.i()) -> str:
    if token not in fake_db_dict:
        raise RuntimeError(f"Can not found by token:{token}")
    return fake_db_dict[token]


class DemoHandler(_Handler):
    @pait(pre_depend_list=[get_user_by_token])
    async def get(self) -> None:
        self.write({"msg": "success"})


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


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

运行代码并执行curl命令后，通过如下输出结果可以看到Pre-Depend能够正常工作。：

curl "http://127.0.0.1:8000/api/demo" --header "token:u12345"{"msg":"success"}curl "http://127.0.0.1:8000/api/demo" --header "token:u123456"{"data":"Can not found by token:u123456"}

Note

1.当使用Pre-Depend时，Pait会先按顺序执行Pre-Depend后再执行路由函数，如果Pre-Depend执行出错则会直接抛错。
2.Pre-Depend绑定是Pait而不是路由函数，这意味着Pre-Depend可以跟随Pait一起复用，详见Pait的复用。

5.不要共享有限的资源

Depend是共享相同逻辑的最佳实现，不过务必注意不要共享有限资源。因为共享的资源是给整个路由函数使用的，这意味着可能会影响到系统的并发数，甚至拖垮整个系统。

Note

有限的资源的种类很多，常见的有限的资源有:线程池，MySQL连接池，Redis连接池等。

由于本节内容与Pait的使用方法无关，所以以Redis的连接举例说明共享有限资源的危害性。

Note

1.最佳的示范用例是MySQL的连接池，不过为了节省代码量，这里采用Redis连接诶吃简要说明共享有限资源的危害。
2.通常情况下是不会直接去获取Redis的连接，Redis也没有暴露出类似的接口，只是execute_command方法的执行逻辑与获取连接类似，所以使用该方法来举例说明。

一个Redis连接只能做一件事，但是Redis本身的设计非常出色，客户端在采用连接池的情况下仍然可以实现高并发。但是如果路由函数的逻辑比较复杂，执行的时间比较久，那么整个服务的并发数就会受限于连接池的数量，如下代码:

import time
from typing import Callable
from flask import Flask, Response, jsonify
from redis import Redis
from pait.app.flask import pait
from pait import field

redis = Redis(max_connections=100)


def get_redis() -> Callable:
    return redis.execute_command


@pait()
def demo(my_redis_conn: Callable = field.Depends.i(get_redis)) -> Response:
    # mock redis cli
    my_redis_conn("info")
    # mock io
    time.sleep(5)
    return jsonify()


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

示例代码中每个路由函数都会先获取Redis连接再执行路由函数逻辑，最终再释放Redis连接。所以Redis连接的使用时间是整个路由函数的运行时间，这意味着如果路由函数的逻辑比较复杂，那么会导致整个服务的并发数受限于Redis连接池的数量。就像示例代码中的demo路由函数一样，demo路由函数会先调用Redis的info命令，然后模拟IO操作睡眠了5秒。这意味着，在获取Redis连接后，Redis的大部分时间都浪费在了等待IO操作上，这是非常糟糕的。

要解决这个问题很简单，只要把共享资源变为共享获取资源的方法就可以了，如下代码:

import time
from typing import Callable
from flask import Flask, Response, jsonify
from redis import Redis
from pait.app.flask import pait
from pait import field

redis = Redis(max_connections=100)


def get_redis() -> Callable:
    return lambda :redis.execute_command


@pait()
def demo(my_redis_conn: Callable = field.Depends.i(get_redis)) -> Response:
    # mock redis cli
    conn = my_redis_conn()
    conn("info")
    del conn
    # mock io
    time.sleep(5)
    return jsonify()


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

该代码有两部分变动，第一部分是第一段高亮代码，该代码的get_redis函数从返回一个Redis连接变为返回一个获取Redis连接的方法，第二部分是第二段高亮代码，从直接调用Redis连接变为先获取Redis连接再调用，最后再释放对Redis连接的占用。这样一来只有使用到了Redis时才会去获取到Redis的连接，系统的并发也就不会很容易的受到Redis连接池影响了。