一次使用 SQLAlchemy 實(shí)現(xiàn)分類以及計(jì)數(shù)的業(yè)務(wù)過程

airborne007 發(fā)布于2019-07-30 16:38 / 1917人閱讀

摘要：在編寫業(yè)務(wù)邏輯代碼的時(shí)候我不幸遇到下面的表結(jié)構(gòu)已經(jīng)將主要邏輯抽離出來了分類分類名稱產(chǎn)品產(chǎn)品名稱分類現(xiàn)在需要實(shí)現(xiàn)的業(yè)務(wù)是返回分類的列表結(jié)果分類這是一個(gè)一對(duì)多的模型一般的笨拙思路就是明眼人一看就知道可以把換成但是根據(jù)這篇文章似乎這樣寫會(huì)有更好的

在編寫業(yè)務(wù)邏輯代碼的時(shí)候, 我不幸遇到下面的表結(jié)構(gòu)(已經(jīng)將主要邏輯抽離出來了):

class Category(Model):
    __tablename__ = "category"
    # 分類ID
    id = Column(Integer, primary_key=True, autoincrement=True)
    # 分類名稱
    name = Column(String(length=255))
    
class Product(Model):
    __tablename__ = "product"
    # 產(chǎn)品 ID
    id = Column(Integer, primary_key=True, autoincrement=True)
    # 產(chǎn)品名稱
    name = Column(String(length=255))
    # 分類 ID
    category_id = Column(Integer)

現(xiàn)在需要實(shí)現(xiàn)的業(yè)務(wù)是返回分類的列表結(jié)果：

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

這是一個(gè)一對(duì)多的模型.
一般的笨拙思路就是:

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()

但是, 根據(jù)這篇文章:[Why is SQLAlchemy count() much slower than the raw query?
](https://stackoverflow.com/que... 似乎這樣寫會(huì)有更好的性能:

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

但是, 稍微有點(diǎn)經(jīng)驗(yàn)的人就會(huì)對(duì)上面的寫法嗤之以鼻, 因?yàn)?b>product_count是放在for category in categorys:里面的, 這意味著如果categorys有成千上萬個(gè), 就要發(fā)出成千上萬個(gè)session.query(), 而數(shù)據(jù)庫請求是在網(wǎng)絡(luò)上的消耗, 請求時(shí)間相對(duì)較長, 有的數(shù)據(jù)庫沒有處理好連接池, 建立連接和斷開連接又是一筆巨大的開銷, 所以 query 的請求應(yīng)該越少越好. 像上面這樣把 query 放到 for 循環(huán)中顯然是不明智的選擇.
于是有了下面一個(gè)請求的版本：

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 函數(shù). 于是改成:

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]

不過這里還有一個(gè)問題, 就是如果先添加一個(gè)Category, 而屬于這個(gè)Category下沒有Product, 那么這個(gè)Category就不會(huì)出現(xiàn)在data里面, 所以join必須改成outerjoin. 即:

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

需求又來了!!!
現(xiàn)在考慮設(shè)計(jì)Product為偽刪除模式, 即添加一個(gè)is_deleted屬性判斷Product是否被刪除.
那么count函數(shù)就不能簡單地count(Product.id), 而是要同時(shí)判斷Product.is_deleted是否為真和Product是否為None, 經(jīng)過悉心研究, 發(fā)現(xiàn)使用func.nullif可以實(shí)現(xiàn)這個(gè)需求,即用下面的寫法:

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 有的時(shí)候還是需要考慮很多東西.