单元测试FastAPI依赖时遭遇类型提示错误的求助

问题描述 #

在FastAPI中使用依赖项prepare_user时，仅单元测试阶段出现Invalid args for response field类型提示错误。正常通过OpenAPI客户端测试应用时可正常返回预期用户信息，其他未使用该依赖项的测试也能正常运行。

代码片段 #

端点代码 #

@router.get(
    '/me/account/user',
    response_model=schemas.User,
)
async def account_user(user: Annotated[schemas.User, Depends(prepare_user)]):
    return user.dict()

依赖项代码 #

async def prepare_user(
    request: Request,
    db: Annotated[AsyncSession, Depends(async_get_db)]
) -> schemas.User:
    """Create a user if it does not exist, otherwise return the user"""

    external_id = request.state.user.claims.get("sub")

    user = await crud.get_user_by_external_id(db=db, external_id=external_id)
    if not user:
        user_in = schemas.UserCreate(external_id=external_id, is_active=True)
        user = await crud.create_user(db=db, user=user_in)

    user = schemas.User.from_orm(user)

    return user

Pytest配置（conftest.py） #

@pytest.fixture(scope='function')
async def db() -> AsyncSession:
    async with async_session() as session:
        yield session

@pytest.fixture(scope='function')
async def normal_client():
    """Fixture for a normal/non-admin user client."""

    async def override_async_get_db(db: AsyncSession) -> AsyncSession:
        try:
            yield db
        finally:
            await db.close()

    app.dependency_overrides[async_get_db] = override_async_get_db

    async def mock_normal_user(request: Request):
        user = B2CUser(
            iss='iss',
            iat=1,
            nbf=2,
            exp=3,
            sub='sub',
            ver='2.0',
            claims={
                'sub': 'sub'
            },
            aud='aud',
            tid='tid',
            access_token='123',
        )
        request.state.user = user
        return user

    app.dependency_overrides[azure_scheme] = mock_normal_user
    async with AsyncClient(app=app, base_url='http://test') as c:
        yield c
    app.dependency_overrides = {}

测试用例 #

async def test_endpoint(normal_client: AsyncClient):
   response = await normal_client.get('/me/account/user')
   assert response.status_code == 200

错误信息 #

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. 

版本信息 #

fastapi             0.97.0
pydantic            1.10.12
pytest              7.4.0
pytest-asyncio      0.21.1

Python              3.11.4

解决方案 #

问题根源 #

测试中覆盖async_get_db的函数override_async_get_db声明了db: AsyncSession参数，但该参数未被Pytest正确注入，反而被FastAPI的依赖注入系统当作需要解析的依赖项处理，导致AsyncSession类型被错误地纳入响应模型校验环节，触发类型提示错误。

具体修改步骤 #

1. 让normal_client fixture依赖db fixture：在normal_client的函数参数中添加db: AsyncSession，确保Pytest将测试用的数据库会话注入进来。
2. 修改override_async_get_db函数：移除该函数的db参数，直接yield注入的测试数据库会话。

修改后的normal_client fixture代码：

@pytest.fixture(scope='function')
async def normal_client(db: AsyncSession):
    """Fixture for a normal/non-admin user client."""

    async def override_async_get_db():
        try:
            yield db
        finally:
            await db.close()

    app.dependency_overrides[async_get_db] = override_async_get_db

    async def mock_normal_user(request: Request):
        user = B2CUser(
            iss='iss',
            iat=1,
            nbf=2,
            exp=3,
            sub='sub',
            ver='2.0',
            claims={
                'sub': 'sub'
            },
            aud='aud',
            tid='tid',
            access_token='123',
        )
        request.state.user = user
        return user

    app.dependency_overrides[azure_scheme] = mock_normal_user
    async with AsyncClient(app=app, base_url='http://test') as c:
        yield c
    app.dependency_overrides = {}

可选优化 #

端点代码中可以直接返回schemas.User实例，FastAPI会自动将其转换为符合response_model要求的格式，无需手动调用.dict()：

@router.get(
    '/me/account/user',
    response_model=schemas.User,
)
async def account_user(user: Annotated[schemas.User, Depends(prepare_user)]):
    return user

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