I am trying to write a middleware in my FastAPI application, so that requests coming to endpoints matching a particular format will be rerouted to a different URL, but I am unable to find a way to do that since request.url
is read-only.
I am also looking for a way to update request headers before rerouting.
Are these things even possible in FastAPI?
Redirection is the best I could do so far:
from fastapi import Request
from fastapi.responses import RedirectResponse
@app.middleware("http")
async def redirect_middleware(request: Request, call_next):
if matches_certain_format(request.url.path):
new_url = create_target_url(request.url.path)
return RedirectResponse(url=new_url)
To change the request
's URL path—in other words, reroute the request to a different endpoint—one can simply modify the request.scope['path']
value inside the middleware, before processing the request, as demonstrated in Option 3 of this answer. If your API endpoints include path parameters (e.g., '/users/{user_id}'
), then you mgiht want to have a look at this answer on how to extract that kind of path from the request
object, and then compare it against a predefined list of routes_to_reroute
, as shown below.
As for updating the request headers, or adding new custom headers to the request, you can follow a similar approach to the one described here, which demonstrates how to modify the request.scope['headers']
value.
If you would like to avoid maintaining a list of routes to reroute and performing checks inside the middleware, you could instead mount a sub-application, which will contain only the routes that require rerouting, and add the middleware to that sub-app, similar to Option 3 of this answer.
from fastapi import FastAPI, Request
app = FastAPI()
routes_to_reroute = ['/']
@app.middleware('http')
async def some_middleware(request: Request, call_next):
if request.url.path in routes_to_reroute:
request.scope['path'] = '/welcome'
headers = dict(request.scope['headers'])
headers[b'custom-header'] = b'my custom header'
request.scope['headers'] = [(k, v) for k, v in headers.items()]
return await call_next(request)
@app.get('/')
async def main():
return 'OK'
@app.get('/welcome')
async def welcome(request: Request):
return {'msg': 'Welcome!', 'headers': request.headers}