How to access route parameter in a middleware? #7226
-
First check
I am trying to add a custom parameter in the route and like to access it in a middleware. 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) |
Beta Was this translation helpful? Give feedback.
Replies: 10 comments
-
|
Middleware is called before router kicks in. You should instead try incorpating the logic into a |
Beta Was this translation helpful? Give feedback.
-
|
thanks for your response. 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 responserequest.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 As a check, I have tags set as and the value of the tag shows in the |
Beta Was this translation helpful? Give feedback.
-
|
What @phy25 meant is that you can't access the endpoint information in the middleware BEFORE the 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 |
Beta Was this translation helpful? Give feedback.
-
|
@Kludex with that, I tried having a breakpoint after but the |
Beta Was this translation helpful? Give feedback.
-
|
@mpdevilleres You're right. Were you able to solve it? I'm debugging it right now. |
Beta Was this translation helpful? Give feedback.
-
|
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 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_handlerAfter we're going to create a custom 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 Following, we have the 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:
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 Possible solution: open a PR on 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) |
Beta Was this translation helpful? Give feedback.
-
|
@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 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. |
Beta Was this translation helpful? Give feedback.
-
|
Btw, if you check the tests on that PR, you'll find an easier solution. Just need the |
Beta Was this translation helpful? Give feedback.
-
|
I think this use case could be easily handled without a middleware, using normal dependencies, here's an example: from fastapi import FastAPI, Depends
class Action:
def __init__(self, action: str):
self.action = action
def __call__(self):
# action parameter available here, do anything you want with audit here
# before the response
print(self.action)
yield
# or after the response
print(self.action)
app = FastAPI()
@app.get("/", dependencies=[Depends(Action("search"))])
def home():
return {"Hello": "World"}It would probably not be a good idea to do that in a middleware as it doesn't have access to any info you put there, as @phy25 says. Or it would require getting the data after the response is created, as @Kludex mentions, but then that requires executing the whole path operation function before being able to access the data in the middleware, and by that point, there's no way to undo any computation that is already done. And that computation could be costly, or that could have saved data records that you wanted to prevent, or could have sent money that you didn't want to be sent. 😅 |
Beta Was this translation helpful? Give feedback.
-
|
@tiangolo thanks for your response... I ended up using dependency injection... :) I am closing this ticket. Thanks |
Beta Was this translation helpful? Give feedback.
I think this use case could be easily handled without a middleware, using normal dependencies, here's an example:
It would probably not be a good idea to do that in a middleware as it doesn't have access to any info you put …