如何将参数传递给 FastAPI 中间件并影响其处理逻辑？

问题描述

我们知道通过 request.state 我们可以将一些自定义数据从中间件的处理前进程传递给处理程序，从而影响其行为，我目前想知道处理程序如何影响中间件之后的逻辑。

我的具体业务场景是，我有一个路由地址（以/api为例），其功能是计算一个动态结果，会消耗比较长的时间，返回比较大的json响应。在操作上，提高效率的一个有效方法是使用一些缓冲区（如redis）来缓存它的结果，这样每次缓存命中时都可以节省计算时间。

由于我的内存有限，我想将 gzip 压缩的字节存储在缓冲区而不是原始 json 流中，这将大大增加我可以处理的缓存量。具体来说，由于响应中有大量的numbers，所以一个响应内容在没有gzip的情况下通常在20MB左右，而压缩后只有1MB左右。这意味着1GB的内存，我不压缩只能缓存50个不同的响应，而压缩我可以缓存1000个，这是一个不容忽视的显着差异。

由于这些要求，我想实现一个功能齐全的 gzip 中间件，但有几个技术上的困惑。首先是我想控制中间件是否压缩，显然，如果响应没有命中缓存而是动态生成，那么它应该被压缩，但相反，它不应该再次压缩，因为它已经已经被压缩过一次。第二个问题是，即使我可以控制中间件不压缩，如何在不需要运行压缩逻辑的情况下将其结果替换为已经压缩的字节？

由于我还不知道如何实现，请原谅我只能提供一些伪代码来说明我的想法。

下面的代码描述了一个比较复杂的响应地址，不包含中间件：

from fastapi import FastAPI,Request
import uvicorn

app = FastAPI()

@app.post("/api")
async def root(request :Request,some_args: int):

    # Using the Fibonacci series to simulate a time-consuming 
    # computational operation
    def fib_recur(n):
    if n <= 1: return n
    return fib_recur(n-1) + fib_recur(n-2)
    
    # In addition to the time-consuming calculation,the size of 
    # the returned content is also large.
    return {"response_content":  [fib_recur(31)] * 10000000 }

if __name__ == '__main__':
    uvicorn.run(app,host='0.0.0.0',port=8080)

我想在添加中间件后实现以下效果：

from fastapi import FastAPI,Request
from fastapi.responses import JSONResponse
from utils import fib_recur,search_for_if_cache_hit
import uvicorn

app = FastAPI()

@app.post("/api")
async def root(request :Request,some_args: int):
    cache_hit_flag,cache_content = search_for_if_cache_hit('/api',some_args)
    if cache_hit_flag == True:
        # Skip Calculation
        response = SomeKindOfResponse(message_content = cache_content)
        response.store['need_gzip'] = False
    else:
        # Calculate normally
        response = JSONResponse( {"response_content":  [fib_recur(31)] * 10000000 } )
        response.store['need_gzip'] = True
    return response

@app.middleware("http")
async def gzip_middleware(request: Request,call_next):
    response = await call_next(request)
    if response.store['need_gzip'] == True:
        response = gzip_handler(response)
    else:
        pass
    return response

if __name__ == '__main__':
    uvicorn.run(app,port=8080)

谢谢！

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）