问题表现

通常我们的基于APIRouter中包含我们的APIRoute的时候，APIRoute中注入我们的依赖项的时候，我们是无法获取到对应的依赖项。如下示例代码：

from fastapi import FastAPI

app = FastAPI()
from fastapi import APIRouter, Depends, Request
from fastapi.routing import APIRoute

router = APIRouter(prefix='/a', tags=["CESIH"])


class TestRouteDepend1:
    def __init__(self, content: str):
        self.content = content

    def __call__(self):
        return "AAAAAAA" + self.content


class TestRouteDepend2:
    def __init__(self, content: str):
        self.content = content

    def __call__(self):
        return "AAAAAAA" + self.content


@router.get("/aaaaa", dependencies=[Depends(TestRouteDepend1(content="你好一号同学")),
                                    Depends(TestRouteDepend2(content="你好二号同学"))])
def ceshi():
    return "我想得到上面【dependencies】的回调结果得不到！！！"


app.include_router(router)

也许你可能会说，那我可以注入到路由函数中，如下代码所示：

@router.get("/aaaaa", dependencies=[Depends(TestRouteDepend1(content="你好一号同学")),
                                    Depends(TestRouteDepend2(content="你好二号同学"))])
def ceshi(yiuhao:str=Depends(TestRouteDepend1(content="你好一号同学")),erhao:str=Depends(TestRouteDepend2(content="你好二号同学"))):
    return "我想得到上面【dependencies】的回调结果:现在可以得到了！！！"+yiuhao+erhao

上面实现确实也没问题，思路也正确。在我自己最新的脚手架，当我通过类的方式定义的时候，我想实现类似JAVA中那种资源依赖注入，并且也希望在对应函数内部能获取到，所以就有了可以实现获取到当我使用如下类似代码的情况也可以获取到对应依赖项结果回调：

@router.get("/aaaaa", dependencies=[Depends(TestRouteDepend1(content="你好一号同学")),
                                    Depends(TestRouteDepend2(content="你好二号同学"))])
def ceshi():
    return "我想得到上面【dependencies】的回调结果得不到！！！"

问题解决思路

首先我们在使用类的方式定义中，我们是通过以下的装饰器上面对应的路由参数，具体实现可以参考前面，以下是大概的代码片段：

def RestRoute(path: str,
              methods: Optional[List[str]] = None,
              response_model: Optional[Type[Any]] = None,
              status_code: Optional[int] = None,
              tags: Optional[List[Union[str, Enum]]] = None,
              dependencies: Optional[Sequence[params.Depends]] = None,
              summary: Optional[str] = None,
              description: Optional[str] = None,
              response_description: str = "Successful Response",
              responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
              deprecated: Optional[bool] = None,
              operation_id: Optional[str] = None,
              response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
              response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
              response_model_by_alias: bool = True,
              response_model_exclude_unset: bool = False,
              response_model_exclude_defaults: bool = False,
              response_model_exclude_none: bool = False,
              include_in_schema: bool = True,
              response_class: Type[Response] = Default(JSONResponse),
              name: Optional[str] = None,
              callbacks: Optional[List[BaseRoute]] = None,
              openapi_extra: Optional[Dict[str, Any]] = None,
              **kwargs: Any):
    def call_func(fun: AnyCallable) -> Callable[[AnyCallable], AnyCallable]:
        route_args = ApiRouteArgs(path=path,
                                  name=name,
                                  status_code=status_code,
                                  methods=methods,
                                  tags=tags,
                                  dependencies=dependencies,
                                  description=description,
                                  summary=summary,
                                  response_description=response_description,
                                  responses=responses,
                                  deprecated=deprecated,
                                  operation_id=operation_id,
                                  response_model_include=response_model_include,
                                  response_model_exclude=response_model_exclude,
                                  response_model_by_alias=response_model_by_alias,
                                  response_model_exclude_unset=response_model_exclude_unset,
                                  response_model_exclude_defaults=response_model_exclude_defaults,
                                  response_model_exclude_none=response_model_exclude_none,
                                  include_in_schema=include_in_schema,
                                  response_class=response_class,
                                  callbacks=callbacks,
                                  openapi_extra=openapi_extra,
                                  response_model=response_model,
                                  **kwargs)

        setattr(fun, '_route_endpoint', fun)
        if not hasattr(fun, '_route_args'):
            setattr(fun, '_route_args', route_args)
        else:
            pass
        return fun

    return call_func

1：思路来源

参考实现透传声明的模型到request的思路，如前篇提到的注入我们的模型到我们的request的代码，如下所示：

def RestParameter(modelss: Type[MODEL_SCHEMA]):
    def call_func(fun: AnyCallable) -> Callable[[AnyCallable], AnyCallable]:
        # 判断是否存在存在路由属性
        if hasattr(fun, '_route_args'):
            route_args: ApiRouteArgs = getattr(fun, '_route_args')

            # 要注入的模型对象依赖项
            def depends_set_model_(request: Request):
                request.state.parameter_model = modelss
                return request

            # 定义依赖项
            parameter: Optional[params.Depends] = Depends(depends_set_model_)
            # 添加依赖项到依赖项列表中
            if route_args.dependencies:
                list(route_args.dependencies).append(parameter)
            else:
                route_args.dependencies = [
                    Depends(depends_set_model_)]

        return fun

    return call_func

在上面的代码中，我们可以通过对生成的route参数信息，然后通过add_api_route传入，如下代码所示：

self.api_router.add_api_route(**route_args.dict(),
                              endpoint=curr_route_endpoint,
                              generate_unique_id_function=Default(generate_unique_id))

而在上面关键点是在于：

 # 要注入的模型对象依赖项
            def depends_set_model_(request: Request):
                request.state.parameter_model = modelss
                return request

我们可以注入一个模型，那我们可以换一个思路，我们也可以注入一个我们的依赖项对象！所以有以下新的装饰器代码：

def RestServicesInject(servicename: str, service=None):
    def call_func(fun: AnyCallable) -> Callable[[AnyCallable], AnyCallable]:
        if hasattr(fun, '_route_args'):
            route_args: ApiRouteArgs = getattr(fun, '_route_args')

            class ServicesContent:
                def __init__(self, content: str):
                    self.content = content

                def __call__(self):
                    return "测试" + self.content

            def depends_set_model_(request: Request,
                                   test: ServicesContent = Depends(ServicesContent(content='你是谁'))):
                setattr(request.state, servicename, test)
                return request

            parameter: Optional[params.Depends] = Depends(depends_set_model_)
            if route_args.dependencies:
                list(route_args.dependencies).append(parameter)
            else:
                route_args.dependencies = [Depends(depends_set_model_)]
        return fun

    return call_func

有了上面的装饰器后，那么我们就可以实现把依赖项分离出来，进行注入，并传递到请求上下文中，如下代码所示：

@RestParameter(modelss=UserModel)
@RestServicesInject(servicename='db_service',service="我现在是在内部定义一个依赖性")
@post("/b/doLogin2ceshi2222/", summary='执行登入接口2')
async def doLoginceshi22222(self, reqsue:Request,suemolde: UserModel):
    # TypeError: Failed to execute 'fetch' on 'Window': Request with GET/HEAD method cannot have body.

    return reqsue.state.db_service

在上面的代码中，通过@RestServicesInject(servicename='db_service',service="我现在是在内部定义一个依赖性")定义一个要注入的依赖项的名和具体的依赖项（注意：这个依赖项我为了演示方便直接在装饰器内部定义了，也就是装饰器内部中ServicesContent类）。经过上面的装饰器装饰之后，我们就可以在对应的路由函数的内部获取到route的路径操作函数中注入的依赖项的结果了。如下图所示：

至此，关于【实现获取Route中路径操作装饰器依赖注入结果回调】相关介绍分享已完成！以上内容分享纯属个人经验，仅供参考！

文笔有限，如有笔误或错误！欢迎批评指正！感谢各位大佬！有什么问题也可以随时交流！

结尾