如何在FastAPI中把API端点的API Credits传递给HTTP中间件？

核心思路 #

给每个API端点函数添加自定义的api_credit属性，在HTTP中间件中通过请求的路由信息读取该属性，进而执行余额扣减逻辑。

实现步骤 #

1. 给端点添加API Credits属性

有两种方式给端点绑定credits值：

方式一：直接给函数赋值（简单直接）

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

@app.post("/myendpoint1")
async def myendpoint1():
    return {"message": "myendpoint"}
# 直接给端点函数添加api_credit属性
myendpoint1.api_credit = 2

@app.post("/myendpoint2")
async def myendpoint2():
    return {"message": "myendpoint2"}
myendpoint2.api_credit = 5

方式二：用装饰器统一管理（更优雅，适合大量端点）

先定义一个装饰器，用来给函数绑定credits值：

def api_credit(credit: int):
    def decorator(func):
        func.api_credit = credit
        return func
    return decorator

然后在端点上使用这个装饰器：

@app.post("/myendpoint1")
@api_credit(2)
async def myendpoint1():
    return {"message": "myendpoint"}

@app.post("/myendpoint2")
@api_credit(5)
async def myendpoint2():
    return {"message": "myendpoint2"}

2. 在中间件中读取并扣减余额

通过request.scope获取当前路由的端点函数，读取其api_credit属性，执行扣减逻辑：

@app.middleware("http")
async def deduct_api_credit(request: Request, call_next):
    # 获取当前路由对象
    route = request.scope.get("route")
    api_credit = 0  # 默认未定义的端点扣0
    
    # 检查端点是否有api_credit属性
    if route and hasattr(route.endpoint, "api_credit"):
        api_credit = route.endpoint.api_credit
        print(f"本次请求需扣减API Credits: {api_credit}")
        
        # 这里实现你的余额扣减逻辑：
        # 1. 从请求中获取用户标识（比如token解析出用户ID）
        # 2. 查询用户当前余额
        # 3. 检查余额是否足够，不足则返回402响应
        # 4. 扣减并更新用户余额
        
        # 示例：模拟余额不足的情况
        # user_balance = get_user_balance(user_id)
        # if user_balance < api_credit:
        #     return JSONResponse(status_code=402, content={"detail": "API Credits余额不足"})
        # update_user_balance(user_id, user_balance - api_credit)
    
    response = await call_next(request)
    return response

注意事项 #

• 如果需要在请求处理前就完成余额检查（避免执行端点逻辑后才发现余额不足），可以在中间件中提前返回错误响应，无需调用call_next(request)。
• 务必保证用户余额操作的原子性，比如使用数据库事务，避免并发请求导致的余额不一致问题。

内容的提问来源于stack exchange，提问作者Moiz Travadi