L�nk?n 2024-06-30 20:56 采纳率: 0%
浏览 127
已结题

FastAPI报错: AsyncSession不是有效Pydantic类型

FastAPI项目,采用了异步SQLAlchemy创建数据库链接,返回一个sqlalchemy.ext.asyncio.session.AsyncSession对象

from sqlalchemy.ext.asyncio import AsyncSession

async def get_db():
    async with async_session() as session:
        yield session

然后在路由函数的参数部分声明依赖,省略了函数执行部分代码:


@router.post('/chat', response_model=None,
             responses=responses, dependencies=[Depends(RateLimiter(times=100, seconds=86400))])
async def chat(new_message: MessageToAI,
                       user_identity: UserIdentity = Depends(get_current_user),
                       db: AsyncSession = Depends(get_db)
                       ) -> Any:

我可以确认如下三点:

  1. 已经在路径装饰中声明response_model=None,使FastAPI不自动创建响应类型;
  2. 路由函数部分声明返回任何数据类型 -> Any;
  3. 函数执行部分仅return状态码,不涉及任何AsyncSession类型。

尽管如此,我还是收到报错:

  File "/app/main.py", line 18, in <module>
    from app.routers import user_router, chat_router, auth_router, page_router
  File "/app/routers/chat_router.py", line 37, in <module>
    @router.post('/chat', response_model=None,
     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
# 省略了部分内容

fastapi.exceptions.FastAPIError: Invalid args for response field! Hint: check that <class 'sqlalchemy.ext.asyncio.session.AsyncSession'> is a valid Pydantic field type. If you are using a return type annotation that is not a valid Pydantic field (e.g. Union[Response, dict, None]) you can disable generating the response model from the type annotation with the path operation decorator parameter response_model=None. Read more: https://fastapi.tiangolo.com/tutorial/response-model/

Process finished with exit code 1

我目前尝试过了通义千问、Kimi、文心一言,给我的建议都是同上述操作1~3,还是不管用。
请给予解答。

7月6日更新:我参照了截止今天所有的回答,都不管用。我在原有路由前增加了一个测试路由,测试路由无报错。见下方代码:

@router.get("/test_db")
async def test_db(db: AsyncSession = Depends(get_db)):
    logging.debug(f"test_db: {db}")
    return {"status": "db_connected"}

@router.post('/chat', response_model=None,
             responses=responses, dependencies=[Depends(RateLimiter(times=100, seconds=86400))])
async def chat_with_ai(new_message: MessageToAI = Body(...),
                       user_identity: UserIdentity = Depends(get_current_user),
                       db: AsyncSession = Depends(get_db),
                       ) -> Any:
    logging.debug('执行Chat')

报错信息:

    @router.post('/chat', response_model=None,
     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
fastapi.exceptions.FastAPIError: Invalid args for response field! Hint: check that <class 'sqlalchemy.ext.asyncio.session.AsyncSession'> is a valid Pydantic field type. If you are using a return type annotation that is not a valid Pydantic field (e.g. Union[Response, dict, None]) you can disable generating the response model from the type annotation with the path operation decorator parameter response_model=None. Read more: https://fastapi.tiangolo.com/tutorial/response-model/
  • 写回答

26条回答 默认 最新

  • 码农阿豪@新空间 新星创作者: 前端开发技术领域 2024-06-30 20:58
    关注
    让阿豪来帮你解答,本回答参考chatgpt3.5编写提供,如果还有疑问可以评论或留言
    根据报错信息和代码片段来看,问题出在响应模型的声明上。 根据FastAPI文档,响应模型可以是任何Pydantic模型。然而,在你的代码中,你设置了`response_model=None`,这将会禁用FastAPI根据函数的返回类型生成默认的响应模型。但是,你的路由函数没有显式地指定响应模型,导致FastAPI无法生成一个有效的响应模型。 为了解决这个问题,你可以采取以下两种方法之一:
    1. 指定适当的响应模型:根据你的业务需求,你可以创建一个Pydantic模型,并将其用作响应模型。然后在你的路由函数中使用response_model参数来指定该模型。例如,假设你的响应模型是名为ResponseModel的Pydantic模型,你可以将路由函数的声明更改为@router.post('/chat', response_model=ResponseModel, ...)
    2. 禁用响应模型的生成:如果你确实不需要自动生成响应模型,你可以在路由函数中使用response_model=None来禁用生成。但是,为了避免报错,你还需要在路由函数的声明中添加response_model_exclude_unset=True参数。这将告诉FastAPI在生成响应模型时排除未设置的字段。例如,你可以将路由函数的声明更改为@router.post('/chat', response_model=None, response_model_exclude_unset=True, ...)。 这是一个示例代码,演示了如何使用上述方法之一解决问题:
    from fastapi import Depends, FastAPI, HTTPException, status
    from sqlalchemy.ext.asyncio import AsyncSession
    from app.schemas import ResponseModel
    app = FastAPI()
    async def get_db():
        async with async_session() as session:
            yield session
    @app.post('/chat', response_model=ResponseModel, response_model_exclude_unset=True)
    async def chat(new_message: MessageToAI, user_identity: UserIdentity = Depends(get_current_user), db: AsyncSession = Depends(get_db)):
        # 执行你的代码逻辑
        return {'message': 'Success'}
    

    请注意,以上代码只是一个示例,你需要根据你的具体情况来进行修改。确保导入所需的模块和自定义模型,并根据你的业务逻辑实现路由函数的代码部分。

    评论

报告相同问题?

问题事件

  • 已结题 (查看结题原因) 7月6日
  • 修改了问题 7月6日
  • 创建了问题 6月30日