如何将指定多表JOIN统计SQL语句转换为SQLAlchemy代码

下面分别给出SQLAlchemy Core和ORM两种方式的实现代码，完全对应你提供的SQL语句逻辑：

1. SQLAlchemy Core 实现 #

先假设你已通过Table类定义对应的数据表结构：

from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, func

metadata = MetaData()

# 定义table1结构
table1 = Table(
    'table1', metadata,
    Column('cust_id', Integer),
    Column('par_id', Integer),
    Column('data', String)
)

# 定义table2结构
table2 = Table(
    'table2', metadata,
    Column('id', Integer, primary_key=True),
    Column('code', String),
    Column('name', String)
)

# 创建数据库引擎并执行查询
engine = create_engine('your_database_url_here')
with engine.connect() as conn:
    query = (
        select(
            func.count(func.distinct(table1.c.cust_id)).label('count'),
            table2.c.code,
            table2.c.name
        )
        .select_from(table1.join(table2, table1.c.par_id == table2.c.id))
        .where(table1.c.data == 'present')
        .group_by(table1.c.par_id)
        .order_by(table2.c.name.asc())
    )
    result = conn.execute(query)
    # 遍历结果
    for row in result:
        print(row.count, row.code, row.name)

2. SQLAlchemy ORM 实现 #

假设你已定义好ORM模型类：

from sqlalchemy import create_engine, Column, Integer, String, func
from sqlalchemy.orm import sessionmaker, declarative_base

Base = declarative_base()

class Table1(Base):
    __tablename__ = 'table1'
    cust_id = Column(Integer)
    par_id = Column(Integer)
    data = Column(String)

class Table2(Base):
    __tablename__ = 'table2'
    id = Column(Integer, primary_key=True)
    code = Column(String)
    name = Column(String)

# 创建会话并执行查询
engine = create_engine('your_database_url_here')
Session = sessionmaker(bind=engine)
session = Session()

query = (
    session.query(
        func.count(func.distinct(Table1.cust_id)).label('count'),
        Table2.code,
        Table2.name
    )
    .join(Table2, Table1.par_id == Table2.id)
    .filter(Table1.data == 'present')
    .group_by(Table1.par_id)
    .order_by(Table2.name.asc())
)

results = query.all()
# 遍历结果
for row in results:
    print(row.count, row.code, row.name)

关键对应点 #

• func.count(func.distinct(...)) 对应SQL中的count(DISTINCT(a.cust_id))
• Core模式下用select_from()配合join()实现内连接，ORM模式直接在query()后调用join()即可
• label('count') 给统计字段设置别名，对应SQL中的as count
• group_by()和order_by()的参数与原SQL逻辑完全匹配

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