如何访问自定义 APIRoute 类中的路由依赖项？

Question

cache我想从自定义 APIRoute 类访问路由级依赖项 ( )。API 是使用路由器和自定义 APIRoute 类定义的。

API路由类

from typing import Callable

from fastapi import Request, Response, Depends
from fastapi.routing import APIRoute

class RecordRequestResponseRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
        
            # need to access cache here 
            response: Response = await original_route_handler(request)
            return response

        return custom_route_handler

路由器API

from fastapi import Response, Depends, APIRouter, BackgroundTasks
import fastapi_plugins
import aioredis

router = APIRouter(route_class=RecordRequestResponseRoute)


@router.get("/users", tags=["users"])
async def match_frame(background_tasks: BackgroundTasks,
                  cache: aioredis.Redis = Depends(fastapi_plugins.depends_redis)):
     return {"success": True, "data": []}

我需要cache在 RecordRequestResponseRoute 类中访问。我尝试使用子依赖项，但这没有帮助。正确的做法是什么？

Answer 1

一段时间以来我一直面临同样的问题，我找到的最佳实现是使用fastapi-utils和基于类的路由。因此，我们可以将依赖项设置为类的属性，并在那里创建任何方法。并像装饰器一样调用它或在路由函数内调用它。

您还可以通过“request.url.path.split("/")[-1]”有效地选择要更改的路由。该字符串返回您的路线。但它也可以通过装饰器来完成，或者您只需在正确的路线内运行此函数即可。

希望它会有用！

代码示例：

# some_route.py
from typing import List

from fastapi import APIRouter, Request, Depends, status, Response
from fastapi_utils.cbv import cbv
from fastapi_utils.inferring_router import InferringRouter

from schemas import UserInfoToken

from utils import get_userinfo
from logger_repo import log_action

router = InferringRouter(tags=["some_tag"])


@cbv(router)
class HardwareRouterClass:
    userinfo: UserInfoToken = Depends(get_userinfo)

    async def user_action_logger(self, request: Request) -> Response:
        user_action = request.url.path.split("/")[-1]

        if user_action == "route_1":
            data = await request.json()

            await log_action(
                user_email=self.userinfo.email,
                device=data["some_key"]
            )

    @router.get("/health")
    async def get_health(self):
        return "OK"

    @router.post("/route_1")
    async def post_route(
        self, request: Request, data: ReadersBody
    ) -> ReadersResponse:
        await self.user_action_logger(request)

        return await your_func(data)