多表关联查询需求：基于ACC_DETIAL关联ORDER/EMAIL/REPORT表统计

单条SQL实现方案 #

核心逻辑是对三张从表分别做分组统计，再通过LEFT JOIN关联主表，把每个acc_id的所有统计字段整合到一行结果中。

SQL语句示例 #

SELECT
    ad.acc_id,
    -- ORDER表统计项
    COALESCE(o.order_count, 0) AS order_count,
    COALESCE(o.total_quantity, 0) AS total_quantity,
    -- EMAIL表统计项
    COALESCE(e.email_count, 0) AS email_count,
    -- REPORT表统计项
    COALESCE(r.report_total, 0) AS report_total,
    COALESCE(r.positive_count, 0) AS positive_count,
    COALESCE(r.negative_count, 0) AS negative_count
FROM ACC_DETIAL ad
LEFT JOIN (
    SELECT
        acc_id,
        COUNT(*) AS order_count,
        SUM(QUANTITY) AS total_quantity
    FROM `ORDER`
    GROUP BY acc_id
) o ON ad.acc_id = o.acc_id
LEFT JOIN (
    SELECT
        acc_id,
        COUNT(*) AS email_count
    FROM EMAIL
    GROUP BY acc_id
) e ON ad.acc_id = e.acc_id
LEFT JOIN (
    SELECT
        acc_id,
        COUNT(*) AS report_total,
        SUM(CASE WHEN TYPE = 'positive' THEN 1 ELSE 0 END) AS positive_count,
        SUM(CASE WHEN TYPE = 'negative' THEN 1 ELSE 0 END) AS negative_count
    FROM REPORT
    GROUP BY acc_id
) r ON ad.acc_id = r.acc_id;

注：用COALESCE是为了避免部分acc_id在从表无数据时返回NULL，统一替换为0。

SQLAlchemy实现方案 #

通过子查询构建各表的统计结果，再用外关联整合到主查询中：

SQLAlchemy代码示例 #

from sqlalchemy import select, func, case
from sqlalchemy.orm import Session
# 假设已定义对应模型类：AccDetial, Order, Email, Report

# 构建ORDER表统计子查询
order_subq = (
    select(
        Order.acc_id,
        func.count().label("order_count"),
        func.sum(Order.quantity).label("total_quantity")
    )
    .group_by(Order.acc_id)
    .subquery()
)

# 构建EMAIL表统计子查询
email_subq = (
    select(
        Email.acc_id,
        func.count().label("email_count")
    )
    .group_by(Email.acc_id)
    .subquery()
)

# 构建REPORT表统计子查询
report_subq = (
    select(
        Report.acc_id,
        func.count().label("report_total"),
        func.sum(case((Report.type == "positive", 1), else_=0)).label("positive_count"),
        func.sum(case((Report.type == "negative", 1), else_=0)).label("negative_count")
    )
    .group_by(Report.acc_id)
    .subquery()
)

# 主查询整合所有统计项
query = (
    select(
        AccDetial.acc_id,
        func.coalesce(order_subq.c.order_count, 0).label("order_count"),
        func.coalesce(order_subq.c.total_quantity, 0).label("total_quantity"),
        func.coalesce(email_subq.c.email_count, 0).label("email_count"),
        func.coalesce(report_subq.c.report_total, 0).label("report_total"),
        func.coalesce(report_subq.c.positive_count, 0).label("positive_count"),
        func.coalesce(report_subq.c.negative_count, 0).label("negative_count")
    )
    .outerjoin(order_subq, AccDetial.acc_id == order_subq.c.acc_id)
    .outerjoin(email_subq, AccDetial.acc_id == email_subq.c.acc_id)
    .outerjoin(report_subq, AccDetial.acc_id == report_subq.c.acc_id)
)

# 执行查询（需传入已创建的Session）
# with Session(engine) as session:
#     results = session.execute(query).fetchall()

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