如何在SqlAlchemy查询中过滤None值并仅追加有效结果？

问题描述 #

使用SQLAlchemy查询Work表，筛选匹配指定order_id且status为"Finished"的记录，获取order_id和document_location，查询代码如下：

loc_list = db.query(Work.order_id, Work.document_location).filter(Work.order_id == order_id, Work.status == "Finished").order_by(desc(Work.last_update)).limit(1).one_or_none()
new_location.append(loc_list)

当Work.status不等于"Finished"时，查询返回None并被追加到new_location列表，最终列表形如：

[None, None, None, None, None, ('Order_id12fb', 'order_12.pdf'),('Order_id13fb', 'order_13.pdf'), ('Order_id14fb', 'order_14.pdf'), ('Order_id15fb', 'order_15.pdf'), ('Order_id16fb', 'order_16.pdf'), None, None]

后续遍历该列表时执行以下逻辑：

for result in new_location:
    order_items = (result.document_location, result.order_id)
return order_items

触发错误：

TypeError: argument of type 'NoneType' is not iterable

希望在SQLAlchemy查询阶段就过滤掉None值，仅将有效的order_id和document_location追加到列表，询问可行的处理方式。

解决方法 #

有几种直接的方式可以在数据处理环节过滤掉无效的None值，避免其进入列表：

1. 追加前判断有效结果 #

在调用append之前检查查询结果是否不为None，仅将有效数据加入列表：

loc_list = db.query(Work.order_id, Work.document_location).filter(Work.order_id == order_id, Work.status == "Finished").order_by(desc(Work.last_update)).limit(1).one_or_none()
if loc_list is not None:
    new_location.append(loc_list)

这种方式无需修改查询语句，仅在数据追加环节做过滤，简单直接。

2. 调整查询语句，优化结果返回 #

如果想尽量在查询阶段规避None，可以将one_or_none()替换为all()，再通过判断结果列表是否为空来决定是否追加：

results = db.query(Work.order_id, Work.document_location).filter(Work.order_id == order_id, Work.status == "Finished").order_by(desc(Work.last_update)).limit(1).all()
if results:
    new_location.append(results[0])

也可以使用first()替代one_or_none()，first()在无匹配结果时同样返回None，但搭配前置判断即可过滤无效值：

loc_list = db.query(Work.order_id, Work.document_location).filter(Work.order_id == order_id, Work.status == "Finished").order_by(desc(Work.last_update)).first()
if loc_list is not None:
    new_location.append(loc_list)

注：one_or_none()和first()的区别在于，当查询返回多条结果时，one_or_none()会抛出异常，而first()仅返回第一条。由于你的代码已经加了limit(1)，二者效果一致。

3. 批量查询优化（多order_id场景） #

如果是批量处理多个order_id，建议一次性查询所有符合条件的记录，减少数据库交互次数，同时自动过滤无匹配的order_id：

# 假设你有一个包含多个order_id的列表
target_order_ids = ["order_id1", "order_id2", ...]
valid_results = db.query(Work.order_id, Work.document_location).filter(Work.order_id.in_(target_order_ids), Work.status == "Finished").order_by(desc(Work.last_update)).all()
new_location.extend(valid_results)

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