FastAPI报错AssertionError：无法为Request指定Depends的解决方法

问题描述 #

使用FastAPI 0.103.1结合Depends、注解及Ray Serve时，遇到错误：AssertionError: Cannot specify 'Depends' for <class 'starlette.requests.Request'>，示例代码如下：

from fastapi import Depends, FastAPI, Request, Response
from fastapi.responses import JSONResponse
from ray import serve
from starlette.middleware.cors import CORSMiddleware
from typing_extensions import Annotated

@asynccontextmanager
async def lifespan(app: FastAPI):
    # do some lifespan stuff like loading configs

app = FastAPI(debug=True, lifespan=lifespan)
app.add_middleware(
    CORSMiddleware,
    allowed_origins=[...],
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["*"]
)

@serve.deployment()
@serve.ingress(app)
class NewAPI:
    @app.post("/login")
    async def login(
        user_email: [Annotated(str, Form()],
        request: Request,
        response: Response
    ):
        # do some processing to create a token
        return token

deployment = NewAPI.bind()
serve.run(deployment)

解决步骤 #

1. 修复语法错误 #

代码中user_email的参数定义存在语法错误，同时缺少Form的导入：

• 错误写法：user_email: [Annotated(str, Form()]
• 正确写法：user_email: Annotated[str, Form()]
• 补充导入：from fastapi import Form

2. 完善lifespan函数 #

lifespan异步上下文管理器必须包含yield语句，否则FastAPI无法正常启动，同时需要补充asynccontextmanager的导入：

• 补充导入：from contextlib import asynccontextmanager
• 在lifespan函数中添加yield

3. 无需降级FastAPI #

FastAPI 0.103.1版本与Ray Serve兼容性正常，不需要回退到旧版本。

修正后的完整代码：

from fastapi import Depends, FastAPI, Request, Response, Form
from fastapi.responses import JSONResponse
from ray import serve
from starlette.middleware.cors import CORSMiddleware
from typing_extensions import Annotated
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # do some lifespan stuff like loading configs
    yield

app = FastAPI(debug=True, lifespan=lifespan)
app.add_middleware(
    CORSMiddleware,
    allowed_origins=["*"],  # 替换为实际允许的域名
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["*"]
)

@serve.deployment()
@serve.ingress(app)
class NewAPI:
    @app.post("/login")
    async def login(
        user_email: Annotated[str, Form()],
        request: Request,
        response: Response
    ):
        # do some processing to create a token
        token = "generated_token"
        return token

deployment = NewAPI.bind()
serve.run(deployment)

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