如何在 FastAPI 中将请求重新路由到不同的 URL/端点？

Question

我正在尝试在我的 FastAPI 应用程序中编写一个中间件，以便到达与特定格式匹配的端点的请求将被重新路由到不同的 URL，但我无法找到一种方法来做到这一点，因为

request.url

是只读的.

我也在寻找一种在重新路由之前更新请求标头的方法。

这些事情在 FastAPI 中可能实现吗？

重定向是迄今为止我能做的最好的事情：

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)

Answer 1

要更改

request

的 URL 路径（换句话说，将请求重新路由到不同的端点），只需在处理请求之前修改中间件内的

request.scope['path']

值，如选项 3 所示这个答案。如果您的 API 端点包含路径参数（例如，

'/users/{user_id}'

），那么您可能需要查看这个答案，了解如何从

request

对象中提取此类路径，然后将其与预定义的路径进行比较。 -定义的

routes_to_reroute

列表，如下所示。

至于更新请求标头，或向请求添加新的自定义标头，您可以遵循与此处描述的类似方法，该方法演示了如何修改

request.scope['headers']

值。

工作示例

如果您想避免维护要重新路由的路由列表并在中间件内部执行检查，您可以挂载一个子应用程序，该子应用程序将仅包含需要重新路由的路由，并将中间件添加到该子应用程序中子应用程序，类似于这个答案的选项3。

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}

Answer 2

是的，FastAPI 应用程序允许 URL 重写。但是，正如您所提到的，request.url 是不可变的，因此您无法更新它。

相反，您可以创建一个新的 URL 对象，并将其传递给具有更新的 URL 和标头的请求对象。下面是一个中间件示例，它同时进行 URL 重写和标头修改：

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)
        headers = request.headers.copy()
        headers['new-header'] = 'new-value'
        new_request = Request(request.scope, request.receive, request.app, request.method, URL(new_url), headers=headers)
        return await call_next(new_request)
    else:
        response = await call_next(request)
        return response