FastAPI中间件无法传递请求至路由处理器问题排查

我在FastAPI中实现了一个TokenDecoder中间件，通过gRPC调用认证服务解码令牌。中间件内部能正常调用认证服务并获取结果，但处理完成后请求无法到达路由处理器，返回422错误，且路由中的print('test', request)无输出。

相关代码 #

中间件代码 #

class TokenDecoder(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):

        print("-" * 60)
        print("middleware girdi")
        print("-" * 60)
        start = time.time()

    async with aio.insecure_channel("127.0.0.1:50051") as channel:
        stub = token_pb2_grpc.TokenServiceStub(channel)

        authorization_header = request.headers.get("Authorization")
        if not authorization_header or not authorization_header.startswith("Bearer "):
            raise HTTPException(
                status_code=401, detail="Authorization token missing")

        token = authorization_header.split(" ")[1]
        grpc_request = token_pb2.UndecodedToken(token=token)

        try:
            response = await stub.GetToken(grpc_request)
            decoded_data = {
                "isValid": response.isValid,
                "userId": response.userId,
                "business": response.business,
                "company": response.company,
            }
            end = time.time()
            total_time = end - start
            print('response insde', response)
            print('request', request)
            print("\nTime Taken to Auth Service	" + str(total_time))
            print("-" * 60)
            # Add decoded_data to the Response object
            request.state.decoded = decoded_data
            print('reste',request.state.decoded)
            result = await call_next(request)
            print('result', result)
            return result
        except grpc.aio.AioRpcError as e:
            print("Error during token validation:", e)
            # Catch the `HTTPException` and return the response from the route
            return await call_next(request)

路由代码 #

@dataprovider.get('/item-list/', tags=["data-provider"])
async def get_itemList(
    request: Request = Depends(TokenDecoder),
    type: Optional[str] = Header(None),
    config: Settings = Depends(get_setting)):
print('test', request)
# print('response', request.state)
return await itemList(company='hktm', type=type, config=config)

main.py 添加中间件代码 #

app.add_middleware(TokenDecoder)
app.add_middleware(Middleware)
app.add_middleware(Logging)
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"]
)

原因分析 #

1. 中间件缩进错误：async with代码块没有缩进在dispatch方法内部，导致这部分核心逻辑完全没有执行，call_next(request)也不会被调用，请求直接被中间件拦截，无法传递到路由。
2. 路由错误依赖中间件：路由中request: Request = Depends(TokenDecoder)的写法错误，TokenDecoder是BaseHTTPMiddleware类型，不能作为依赖注入使用。FastAPI会尝试将其解析为依赖，引发参数解析错误，这也是返回422状态码的直接原因。
3. 异常处理逻辑不合理：当gRPC调用出错时，直接返回await call_next(request)，此时认证未完成，请求不应该继续传递到路由，而且这种处理方式会导致错误的请求流程。

解决方案 #

1. 修复中间件缩进 #

将async with代码块缩进，确保它属于dispatch方法的内部逻辑：

class TokenDecoder(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        print("-" * 60)
        print("middleware girdi")
        print("-" * 60)
        start = time.time()

        # 修复缩进：将这部分代码放入dispatch方法内
        async with aio.insecure_channel("127.0.0.1:50051") as channel:
            stub = token_pb2_grpc.TokenServiceStub(channel)

            authorization_header = request.headers.get("Authorization")
            if not authorization_header or not authorization_header.startswith("Bearer "):
                raise HTTPException(
                    status_code=401, detail="Authorization token missing")

            token = authorization_header.split(" ")[1]
            grpc_request = token_pb2.UndecodedToken(token=token)

            try:
                response = await stub.GetToken(grpc_request)
                decoded_data = {
                    "isValid": response.isValid,
                    "userId": response.userId,
                    "business": response.business,
                    "company": response.company,
                }
                end = time.time()
                total_time = end - start
                print('response insde', response)
                print('request', request)
                print("\nTime Taken to Auth Service	" + str(total_time))
                print("-" * 60)
                request.state.decoded = decoded_data
                print('reste',request.state.decoded)
                result = await call_next(request)
                print('result', result)
                return result
            except grpc.aio.AioRpcError as e:
                print("Error during token validation:", e)
                # 认证失败时返回错误响应，而非继续传递请求
                return JSONResponse(status_code=401, detail="Token validation failed")

2. 修正路由依赖 #

移除路由中错误的Depends(TokenDecoder)，直接声明request: Request即可：

@dataprovider.get('/item-list/', tags=["data-provider"])
async def get_itemList(
    request: Request,
    type: Optional[str] = Header(None),
    config: Settings = Depends(get_setting)):
    print('test', request)
    # 可以直接使用request.state.decoded获取中间件注入的数据
    print('decoded data', request.state.decoded)
    return await itemList(company='hktm', type=type, config=config)

3. 优化异常处理 #

gRPC调用失败时，直接返回错误响应，避免无效请求进入路由。

内容的提问来源于stack exchange，提问作者Furkan YIlmaZ