Fastapi: How to access route parameter in a middleware?

[mpdevilleres]

First check

• [X] I added a very descriptive title to this issue.
• [X] I used the GitHub search to find a similar issue and didn't find it.
• [X] I searched the FastAPI documentation, with the integrated search.
• [X] I already searched in Google "How to X in FastAPI" and didn't find any information.
• [X] I already read and followed all the tutorial in the docs and didn't find an answer.
• [X] I already checked if it is not related to FastAPI but to Pydantic.
• [X] I already checked if it is not related to FastAPI but to Swagger UI.
• [X] I already checked if it is not related to FastAPI but to ReDoc.
• [X] After submitting this, I commit to one of:
  
  
  
  • Read open issues with questions until I find 2 issues where I can help someone and add a comment to help there.
  
  
  • I already hit the "watch" button in this repository to receive notifications and I commit to help at least 2 people that ask questions in the future.
  
  
  • Implement a Pull Request for a confirmed bug.
  
  

I am trying to add a custom parameter in the route and like to access it in a middleware.
eg.

from fastapi import FastAPI
app = FastAPI()

@app.get("/", action=['search'])  # I am interested in accessing the `action` parameter in a middleware
def read_root():
    return {"Hello": "World"}

Here's what I have done so far.

from typing import Callable, List, Any

import uvicorn
from fastapi.routing import APIRoute
from starlette.requests import Request
from starlette.responses import Response

from starlette.middleware.base import (
    BaseHTTPMiddleware,
    RequestResponseEndpoint
)

from fastapi import FastAPI, APIRouter


class AuditHTTPMiddleware(BaseHTTPMiddleware):
    # I want the `action` parameter from the route to be accessible in this middleware
    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        response = await call_next(request)
        return response


class CustomAPIRoute(APIRoute):
    def __init__(self, path: str, endpoint: Callable, *, action: List[str] = None, **kwargs) -> None:
        super(CustomAPIRoute, self).__init__(path, endpoint, **kwargs)
        self.action = action


class CustomAPIRouter(APIRouter):
    def __init__(self):
        super(CustomAPIRouter, self).__init__()
        self.route_class = CustomAPIRoute

    def add_api_route(self, path: str, endpoint: Callable, *, action: List[str] = None, **kwargs) -> None:
        route = self.route_class(path=path, endpoint=endpoint, action=action, **kwargs)
        self.routes.append(route)

    def api_route(self, path: str, *, action: List[str] = None, **kwargs) -> Callable:
        def decorator(func: Callable) -> Callable:
            self.add_api_route(path, func, action=action, **kwargs)
            return func

        return decorator

    def get(self, path: str, *, action: List[str] = None, **kwargs) -> Callable:
        return self.api_route(path, action=action, **kwargs)


router = CustomAPIRouter()


@router.get('/', action=['Search'], tags=["Current tag"])
def get_data():
    return {}


if __name__ == '__main__':
    app = FastAPI()

    app.add_middleware(AuditHTTPMiddleware)
    app.include_router(prefix='', router=router)
    uvicorn.run(app, host="0.0.0.0", port=9002)

[phy25]

Middleware is called before router kicks in. You should instead try incorpating the logic into a Depends or reading query params from the original request yourself.

[mpdevilleres]

@phy25

thanks for your response.
I am not sure if I understood what you mean called before as I am able to get request and response in the middleware.

Additionally I found something interesting just now.

that inside the middleware you will be able to access the list of all routes with their corresponding parameters.

class AuditHTTPMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        # BREAKPOINT HERE
        response = await call_next(request)
        return response

request.scope['router'].routes[-1]
"""
action = {NoneType} None
body_field = {NoneType} None
callbacks = {NoneType} None
dependant = {Dependant} <fastapi.dependencies.models.Dependant object at 0x11033fe80>
dependencies = {list: 0} []
dependency_overrides_provider = {FastAPI} <fastapi.applications.FastAPI object at 0x10fa5c520>
deprecated = {NoneType} None
description = {str} ''
include_in_schema = {bool} True
methods = {set: 1} {'GET'}
name = {str} 'get_data'
operation_id = {NoneType} None
param_convertors = {dict: 0} {}
path = {str} '/'
path_format = {str} '/'
path_regex = {Pattern} re.compile('^/$')
response_class = {type} <class 'starlette.responses.JSONResponse'>
response_description = {str} 'Successful Response'
response_field = {NoneType} None
response_fields = {dict: 0} {}
response_model = {NoneType} None
response_model_by_alias = {bool} True
response_model_exclude = {NoneType} None
response_model_exclude_defaults = {bool} False
response_model_exclude_none = {bool} False
response_model_exclude_unset = {bool} False
response_model_include = {NoneType} None
responses = {dict: 0} {}
secure_cloned_response_field = {NoneType} None
status_code = {int} 200
summary = {NoneType} None
tags = {list: 1} ['Current tag']
unique_id = {str} 'get_data__get'
"""

But not sure why the action is None

As a check, I have tags set as ['Current tag']

@router.get('/', action=['Search'], tags=["Current tag"])

and the value of the tag shows in the request.scope['router'].routes[-1]

[Kludex]

What @phy25 meant is that you can't access the endpoint information in the middleware BEFORE the call_next(). Here you have an example on what he meant and explaining why are you able to access the information on the previous post.

from starlette.middleware.base import BaseHTTPMiddleware
from fastapi import FastAPI, Request

class AuditHTTPMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if router := request.scope.get('router'):  # Maybe you should add "== None" to visualize the message :)
            print("I can't access before the endpoint call!")
        response = await call_next(request)
        if request.scope.get('router'):
            print("I have 'router' info now!")
        return response

app = FastAPI()
app.add_middleware(AuditHTTPMiddleware)

@app.get('/')
def home():
    pass

[mpdevilleres]

@Kludex
thanks for clarification. I understand the point now.

with that, I tried having a breakpoint after call_next and inspect the value of the route

request.scope['router'].routes[-1]
'''
action = {NoneType} None
body_field = {NoneType} None
callbacks = {NoneType} None
dependant = {Dependant} <fastapi.dependencies.models.Dependant object at 0x109d24b50>
dependencies = {list: 0} []
dependency_overrides_provider = {FastAPI} <fastapi.applications.FastAPI object at 0x109444580>
deprecated = {NoneType} None
description = {str} ''
include_in_schema = {bool} True
methods = {set: 1} {'GET'}
name = {str} 'get_data'
operation_id = {NoneType} None
param_convertors = {dict: 0} {}
path = {str} '/'
path_format = {str} '/'
path_regex = {Pattern} re.compile('^/$')
response_class = {type} <class 'starlette.responses.JSONResponse'>
response_description = {str} 'Successful Response'
response_field = {NoneType} None
response_fields = {dict: 0} {}
response_model = {NoneType} None
response_model_by_alias = {bool} True
response_model_exclude = {NoneType} None
response_model_exclude_defaults = {bool} False
response_model_exclude_none = {bool} False
response_model_exclude_unset = {bool} False
response_model_include = {NoneType} None
responses = {dict: 0} {}
secure_cloned_response_field = {NoneType} None
status_code = {int} 200
summary = {NoneType} None
tags = {list: 1} ['Current tag']
unique_id = {str} 'get_data__get'
'''

but the action is set to None

[Kludex]

@mpdevilleres You're right. Were you able to solve it? I'm debugging it right now.

[Kludex]

Ok. I'm able to answer now. :)

I'll be using my own use case to explain what's happening to your code and mine as well.

So, we're going to define a custom APIRoute, which we want to have a custom field called permissions (or your action):

class CustomAPIRoute(APIRoute):
    def __init__(
        self,
        path: str,
        endpoint: Callable,
        *,
        permissions: Optional[Iterable[str]] = None,
        **kwargs
    ) -> None:
        super().__init__(path, endpoint, **kwargs)
        self.permissions = permissions

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

        async def custom_route_handler(request: Request) -> Response:
            request.state.permissions = self.permissions
            response = await original_route_handler(request)
            return response

        return custom_route_handler

After we're going to create a custom APIRouter that has as default route, the one we define above. Let's see it:

class CustomAPIRouter(APIRouter):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.route_class = CustomAPIRoute

    def add_api_route(
        self,
        path: str,
        endpoint: Callable,
        *,
        permissions: Optional[Iterable[str]] = None,
        **kwargs
    ) -> None:
        route = self.route_class(
            path=path, endpoint=endpoint, permissions=permissions, **kwargs
        )
        self.routes.append(route)

    def api_route(
        self, path: str, *, permissions: Optional[Iterable[str]] = None, **kwargs
    ) -> Callable:
        def decorator(func: Callable) -> Callable:
            self.add_api_route(path, func, permissions=permissions, **kwargs)
            return func

        return decorator

    def get(
        self, path: str, *, permissions: Optional[Iterable[str]] = None, **kwargs
    ) -> Callable:
        return self.api_route(path, permissions=permissions, **kwargs)

On this case, we're only defining the get method, and yes, we need to define it because we want the custom field permissions, the original get from APIRouter doesn't have this field, hence we'd receive an error.

Following, we have the app and router instances:

app = FastAPI()
router = CustomAPIRouter()

@router.get("/", permissions=["Potato", "Tomato"])
def home(request: Request):
    return request.state.permissions

app.include_router(router)

What happens here is:

1. FastAPI app is created.
2. CustomAPIRouter is created.
3. CustomAPIRoute is created and saved in the routes attribute in the CustomAPIRouter.
4. FastAPI app includes the router router.

The problem is in the step 4. If you check the implementation, you'll see:

    def include_router(
        self,
        router: routing.APIRouter,
        *,
        prefix: str = "",
        tags: Optional[List[str]] = None,
        dependencies: Optional[Sequence[Depends]] = None,
        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
        default_response_class: Optional[Type[Response]] = None,
    ) -> None:
        self.router.include_router(
            router,
            prefix=prefix,
            tags=tags,
            dependencies=dependencies,
            responses=responses or {},
            default_response_class=default_response_class
            or self.default_response_class,
        )

Which tries to include our router inside the application router. But if you follow the path to the inner include_router(), you'll see the field permission/action or wherever you want to create is not send to your CustomAPIRoute!

Possible solution: open a PR on routing.py to update add_api_route() so it accepts unkown attributes. @tiangolo Would you accept a PR with it? Do we want that? Alternatively we can check if the APIRoute class is actually a subclass of APIRoute and giving the full responsability to the custom APIRoute.

Full example:

from typing import Callable, Iterable, Optional

from fastapi import FastAPI, APIRouter, Response, Request
from fastapi.routing import APIRoute

app = FastAPI()


class CustomAPIRoute(APIRoute):
    def __init__(
        self,
        path: str,
        endpoint: Callable,
        *,
        permissions: Optional[Iterable[str]] = None,
        **kwargs
    ) -> None:
        super().__init__(path, endpoint, **kwargs)
        self.permissions = permissions

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

        async def custom_route_handler(request: Request) -> Response:
            request.state.permissions = self.permissions
            response = await original_route_handler(request)
            return response

        return custom_route_handler


class CustomAPIRouter(APIRouter):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.route_class = CustomAPIRoute

    def add_api_route(
        self,
        path: str,
        endpoint: Callable,
        *,
        permissions: Optional[Iterable[str]] = None,
        **kwargs
    ) -> None:
        route = self.route_class(
            path=path, endpoint=endpoint, permissions=permissions, **kwargs
        )
        self.routes.append(route)

    def api_route(
        self, path: str, *, permissions: Optional[Iterable[str]] = None, **kwargs
    ) -> Callable:
        def decorator(func: Callable) -> Callable:
            self.add_api_route(path, func, permissions=permissions, **kwargs)
            return func

        return decorator

    def get(
        self, path: str, *, permissions: Optional[Iterable[str]] = None, **kwargs
    ) -> Callable:
        return self.api_route(path, permissions=permissions, **kwargs)


router = CustomAPIRouter()


@router.get("/", permissions=["Potato", "Tomato"])
def home(request: Request):
    return request.state.permissions


app.include_router(router)

[mpdevilleres]

@Kludex I appreciate you taking time with this issue. and I believe its a valid use case specially when we want to to have middlewares.

I kinda did the same as you have implemented it, I override APIRoute and APIRouter. but can't still use a middleware as the value of action/permission in the middleware is None, I did some digging and saw the at include_api_routes in the applications.FastAPI is not recognizing my changes. so I end up just have the APIRoute without a middleware.

and looking at your PR 1917 might be the solution to my use case of having a middleware. I will test it out and share my findings.

[Kludex]

Btw, if you check the tests on that PR, you'll find an easier solution.

Just need the CustomAPIRoute, not the router.