FastAPI中间件如何更新/修改请求头和查询参数？

Question

我正在尝试为 FastAPI 项目编写一个中间件，该项目在某些特殊情况下操纵请求

headers

和/或

query

参数。

我已经设法捕获并修改了中间件中的请求对象，但似乎即使我修改了传递给中间件的请求对象，服务端点的函数也会收到原始的、未修改的请求。

这是我的实现的简化版本：

from fastapi import FastAPI, Request
from starlette.datastructures import MutableHeaders, QueryParams
from starlette.middleware.base import BaseHTTPMiddleware


class TestMiddleware(BaseHTTPMiddleware):


    def __init__(self, app: FastAPI):
        super().__init__(app)

    
    def get_modified_query_params(request: Request) -> QueryParams:

        pass  ## Create and return new query params


    async def dispatch(
        self, request: Request, call_next, *args, **kwargs
    ) -> None:
        
        # Check and manipulate the X-DEVICE-TOKEN if required
        header_key = "X-DEVICE-INFo"
        new_header_value = "new device info"

        new_header = MutableHeaders(request._headers)
        new_header[header_key] = new_header_value

        request._headers = new_header

        request._query_params = self.get_modified_query_params(request)

        print("modified headers  =>", request.headers)
        print("modified params  =>", request.query_params)

        return await call_next(request)

即使我在上面的打印语句中看到了更新的值，当我尝试在服务于端点的函数中打印请求对象时，我看到了请求的原始值。

我错过了什么？

Answer 1

您需要更新

request.scope['headers']

，如this answer中所述。这样，您可以添加新的自定义标头，也可以修改现有的标头。以类似的方式，通过更新

request.scope['query_string']

，您可以修改现有的查询参数，以及添加新的。

工作示例

from fastapi import FastAPI, Request
from urllib.parse import urlencode

app = FastAPI()

@app.middleware('http')
async def some_middleware(request: Request, call_next):
    # update request headers
    headers = dict(request.scope['headers'])
    headers[b'custom-header'] = b'my custom header'
    request.scope['headers'] = [(k, v) for k, v in headers.items()]
    
    # update request query parameters
    q_params = dict(request.query_params)
    q_params['custom-q-param'] = 'my custom query param'
    request.scope['query_string'] = urlencode(q_params).encode('utf-8')
        
    return await call_next(request)


@app.get('/')
async def main(request: Request):
    return {'headers': request.headers, 'q_params': request.query_params}