FastAPI接口返回空响应排查求助（数据库存在数据）

我用SQLAlchemy和Alembic搭建了FastAPI应用，但某一接口始终返回空响应——尽管数据库中存在对应数据。排查后怀疑问题出在camera_service.py和camera_repository.py文件中。该接口负责按指定过滤器查询数据库并返回结果，虽然已实现数据查询与响应对象转换逻辑，但返回的摄像头列表始终为空，count值固定为0，请求协助排查解决，让接口返回预期数据。

响应示例 #

{
  "cameras": [],
  "count": 0
}

相关代码 #

接口路由代码 #

@router.get('/camera/v1', response_model=AllCamerasResponse, summary="按过滤器查询摄像头")
async def search_cameras(
    cam_name: str = None,
    model: str = None,
    is_live_on: bool = False,
    is_live_audio_on: bool = False,
):
    query_filters = {
        "cam_name": cam_name,
        "model": model,
        "is_live_on": is_live_on,
        "is_live_audio_on": is_live_audio_on
    }
    return await camera_service.search_cameras(**query_filters)

数据库仓库代码 #

class CameraRepository:
    async def get_all_cameras(self, **query_filters):
        async with get_db() as session:
            stmt = select(CameraDB)
            stmt = select(CameraDB).filter_by(**query_filters)
            result = await session.execute(stmt)
            return result.scalars().all()

1. 过滤参数默认值导致的核心问题 #

接口中is_live_on和is_live_audio_on的默认值设为False，当用户不传入这两个参数时，filter_by会强制匹配is_live_on=False的记录，但数据库中对应字段可能是NULL或True，导致无匹配结果。

修复方式：仅在参数有明确传入值时才添加过滤条件，避免默认值干扰：

@router.get('/camera/v1', response_model=AllCamerasResponse, summary="按过滤器查询摄像头")
async def search_cameras(
    cam_name: str = None,
    model: str = None,
    is_live_on: bool = None,  # 改为None，无传入时不参与过滤
    is_live_audio_on: bool = None,
):
    query_filters = {}
    if cam_name is not None:
        query_filters["cam_name"] = cam_name
    if model is not None:
        query_filters["model"] = model
    if is_live_on is not None:
        query_filters["is_live_on"] = is_live_on
    if is_live_audio_on is not None:
        query_filters["is_live_audio_on"] = is_live_audio_on
    return await camera_service.search_cameras(**query_filters)

2. 清理冗余查询语句 #

仓库代码中stmt = select(CameraDB)属于冗余代码，虽不影响功能，但可删除以保持简洁：

class CameraRepository:
    async def get_all_cameras(self, **query_filters):
        async with get_db() as session:
            stmt = select(CameraDB).filter_by(**query_filters)
            result = await session.execute(stmt)
            return result.scalars().all()

3. 验证服务层转换逻辑 #

若上述修改后仍返回空，需检查camera_service.search_cameras的处理逻辑：

• 确认服务层未额外过滤仓库返回的结果
• 确保AllCamerasResponse的count字段是基于查询结果长度计算，而非硬编码为0

示例正确的服务层逻辑：

class CameraService:
    def __init__(self, repo: CameraRepository):
        self.repo = repo

    async def search_cameras(self, **query_filters):
        cameras_db = await self.repo.get_all_cameras(**query_filters)
        # 转换为响应模型对象
        cameras = [CameraResponse.from_orm(cam) for cam in cameras_db]
        return AllCamerasResponse(cameras=cameras, count=len(cameras))

4. 确认数据库与数据匹配 #

• 检查FastAPI配置的数据库连接，确保和你查看数据的数据库是同一个（避免开发/测试库混淆）
• 手动执行SQL查询验证过滤条件：

  SELECT * FROM camera WHERE cam_name = '测试名称' AND model = '测试型号';
  

  替换为实际参数值，确认数据库中存在符合条件的记录。

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