如何优化SQLAlchemy查询结果按字段分组的循环代码

1. 使用collections.defaultdict简化字典操作 #

原代码中需要手动判断分类是否存在，用defaultdict可以自动初始化空列表，省去判断逻辑，让循环代码更简洁：

from collections import defaultdict

skills = db.query(model.Skill).order_by(model.Skill.category).all()

_categories = defaultdict(list)
for skill in skills:
    _categories[skill.category].append({"name": skill.name, "img_path": skill.img_path})

# 转换为目标格式的列表
result = [{"category": cat, "skills": skills_list} for cat, skills_list in _categories.items()]

2. 利用SQLAlchemy在数据库层面分组（适合大数据量场景） #

如果数据量较大，把分组逻辑交给数据库处理能减少应用层的内存消耗，结合SQL的聚合函数直接构造需要的结构：

from sqlalchemy import func

# 构造单个技能的字典结构
skill_struct = func.json_build_object(
    'name', model.Skill.name,
    'img_path', model.Skill.img_path
)

# 按category分组并聚合技能数据
query = db.query(
    model.Skill.category,
    func.array_agg(skill_struct).label('skills')
).group_by(model.Skill.category).order_by(model.Skill.category)

# 直接转换为目标格式
result = [row._asdict() for row in query.all()]

这种方式将分组计算转移到数据库，应用层仅需转换结果格式，性能表现更优。

3. 使用itertools.groupby实现分组 #

由于原查询已经按category排序，itertools.groupby可以直接对连续相同分类的项进行分组，代码更具Python风格：

from itertools import groupby
from operator import attrgetter

skills = db.query(model.Skill).order_by(model.Skill.category).all()

result = []
for category, group in groupby(skills, key=attrgetter('category')):
    result.append({
        "category": category,
        "skills": [{"name": s.name, "img_path": s.img_path} for s in group]
    })

groupby会自动将同分类的连续项归为一组，配合排序后的查询结果，代码简洁直观。

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