How to access route dependencies in custom APIRoute class?

user_6826059 0
Viewed 792

I want to access a route-level dependency (cache) from a custom APIRoute class. The API is defined using a router and a custom APIRoute class.

APIRoute class

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

Router 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": []}

I need to access cache in RecordRequestResponseRoute class. I tried using sub dependencies but that did not help. What will be the correct way to do it?

fastapi middleware python
1 Answers

I've been facing the same issue for some time and the best implementation I've came to is using fastapi-utils and class-based routing. Thus we can set up our dependencies as attributes of the class and create any method there. And call it like decorator or just inside route function.

Also you can efficiently choose which route will be changed by "request.url.path.split("/")[-1]". The string returns you the route. But it can also be done via decorator or you just run this func inside proper route.

Hope it will be useful!

Code example:

# 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)
user_142299650
Related