一次使用 SQLAlchemy 实现分类以及计数的业务过程

摘要：在编写业务逻辑代码的时候我不幸遇到下面的表结构已经将主要逻辑抽离出来了分类分类名称产品产品名称分类现在需要实现的业务是返回分类的列表结果分类这是一个一对多的模型一般的笨拙思路就是明眼人一看就知道可以把换成但是根据这篇文章似乎这样写会有更好的

在编写业务逻辑代码的时候, 我不幸遇到下面的表结构(已经将主要逻辑抽离出来了):

class Category(Model):
    __tablename__ = "category"
    # 分类ID
    id = Column(Integer, primary_key=True, autoincrement=True)
    # 分类名称
    name = Column(String(length=255))
    
class Product(Model):
    __tablename__ = "product"
    # 产品 ID
    id = Column(Integer, primary_key=True, autoincrement=True)
    # 产品名称
    name = Column(String(length=255))
    # 分类 ID
    category_id = Column(Integer)

现在需要实现的业务是返回分类的列表结果：

[
    {
        "id": 1,
        "name": "分类1",
        "product_count": 1
    },
    ...
]

这是一个一对多的模型.
一般的笨拙思路就是:

data = []
categorys = Category.query.all()
for category in categorys:
    product_count = len(Product.query.filter(Product.category_id == category.id).all())
    data.append({
        "id": category.id,
        "name": category.name,
        "product_count": product_count
    })

明眼人一看就知道可以把len(Product.query.filter(Product.category_id == category.id).all())换成:

product_count = Product.query.filter(Product.category_id == category.id).count()

但是, 根据这篇文章:[Why is SQLAlchemy count() much slower than the raw query?
](https://stackoverflow.com/que... 似乎这样写会有更好的性能:

from sqlalchemy import func
session.query(func.count(Product.id)).filter(Product.category_id == category.id).scalar()

但是, 稍微有点经验的人就会对上面的写法嗤之以鼻, 因为product_count是放在for category in categorys:里面的, 这意味着如果categorys有成千上万个, 就要发出成千上万个session.query(), 而数据库请求是在网络上的消耗, 请求时间相对较长, 有的数据库没有处理好连接池, 建立连接和断开连接又是一笔巨大的开销, 所以 query 的请求应该越少越好. 像上面这样把 query 放到 for 循环中显然是不明智的选择.
于是有了下面一个请求的版本：

result = db.session.query(Product, Category) 
    .filter(Product.category_id == Category.id)
    .order_by(Category.id).all()
id_list = []
data = []
for product, category in result:
    if category and product:
        if category.id not in id_list:
            id_list.append(category.id)
            data.append({
                "id": category.id,
                "name": category.name,
                "product_count": 0
            })
        idx = id_list.index(category.id)
        data[idx]["product_count"] += 1  

这样的写法十分难看, 而且同样没有合理利用 SQLAlchemy 的 count 函数. 于是改成:

product_count = func.count(Product.id).label("count")
results = session.query(Category, product_count) 
    .join(Product, Product.category_id == Category.id) 
    .group_by(Category).all()
data = [
    {
        "id": category.id,
        "name": category.name,
        "product_count": porduct_count
    } for category, product_count in results]

不过这里还有一个问题, 就是如果先添加一个Category, 而属于这个Category下没有Product, 那么这个Category就不会出现在data里面, 所以join必须改成outerjoin. 即:

results = session.query(Category, product_count) 
    .outerjoin(Product, Product.category_id == Category.id) 
    .group_by(Category).all()

需求又来了!!!
现在考虑设计Product为伪删除模式, 即添加一个is_deleted属性判断Product是否被删除.
那么count函数就不能简单地count(Product.id), 而是要同时判断Product.is_deleted是否为真和Product是否为None, 经过悉心研究, 发现使用func.nullif可以实现这个需求,即用下面的写法:

product_count = func.count(func.nullif(Product.is_deleted.is_(False), False)).label("count")
results = session.query(Category, product_count) 
    .join(Product, Product.category_id == Category.id) 
    .group_by(Category).all()
data = [
    {
        "id": category.id,
        "name": category.name,
        "product_count": porduct_count
    } for category, product_count in results]

可见使用 ORM 有的时候还是需要考虑很多东西.