FastAPI+SQLAlchemy按BookId聚合嵌套数据返回空问题排查
问题分析与修复
你的代码存在以下关键错误,导致返回空响应:
1. 查询未过滤指定book_id
当前SQL查询拉取全表数据,未添加WHERE book_id = :id的筛选条件,后续处理的是所有数据而非目标book_id的数据。
2. 分组逻辑错误
- 用
book_name作为分组key,但需求是按book_id聚合,book_id是唯一标识,用它分组更准确(避免同名book的情况)。 - 循环内每次迭代都重新生成
result,属于冗余操作,应在分组完成后统一生成结果。
3. 最终返回逻辑完全无效
最后的while groups[book_id] == id逻辑错误:groups[book_id]是列表,id是整数,两者永远不可能相等,循环不会执行,函数无返回值,自然返回空响应。
4. 未清理嵌套字段
未将book_id、book_name从writing_items子项中移除,会导致子项包含重复的外层字段。
修正后的代码实现
def get_one_group(id: int, db: Session): # 1. 添加WHERE条件过滤指定book_id的数据 data = db.execute( 'SELECT id, book_id, book_name, writing_test, writing_test_description ' 'FROM public."tbl_study_content-writing_test_name" ' 'WHERE book_id = :book_id;', {"book_id": id} ) test_records = data.mappings().all() if not test_records: return [] # 无数据时返回空列表 # 2. 按book_id分组(已过滤,实际仅一个分组,保留通用逻辑) from collections import defaultdict groups = defaultdict(lambda: {"book_id": None, "book_name": None, "writing_items": []}) for test in test_records: book_id = test['book_id'] # 初始化外层的book_id和book_name if groups[book_id]["book_id"] is None: groups[book_id]["book_id"] = book_id groups[book_id]["book_name"] = test['book_name'] # 构造子项,移除外层不需要的字段 item = { "id": test['id'], "writing_test": test['writing_test'], "writing_test_description": test['writing_test_description'] } groups[book_id]["writing_items"].append(item) # 转换为需求的数组格式 result = list(groups.values()) return result
额外优化建议(使用SQLAlchemy ORM)
用ORM替代原生SQL,代码更简洁易维护:
def get_one_group(id: int, db: Session): # ORM查询指定book_id的所有记录 test_records = db.query(Writing_Test_Name).filter(Writing_Test_Name.book_id == id).all() if not test_records: return [] # 直接构造目标结构 result = { "book_id": id, "book_name": test_records[0].book_name, "writing_items": [ { "id": t.id, "writing_test": t.writing_test, "writing_test_description": t.writing_test_description } for t in test_records ] } return [result] # 保持需求的数组格式
内容的提问来源于stack exchange,提问作者trencherman
相关产品推荐
相关产品推荐

