如何用SQLAlchemy的having与func.count优化关联查询避免两次查询

我有两张表：order_hist（父表）和private_trade（子表），表结构定义如下：

op.create_table(
        'order_hist',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('algo_order_id', sa.Integer, sa.ForeignKey("algo_order.id")),
        ... other fields ...
        sa.Column('status', sa.String(), index=True, nullable=False),
    )
    
op.create_table(
        'private_trade',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('order_id', sa.String(), sa.ForeignKey("order_hist.order_id")),
        sa.Column('hedged', sa.Boolean(), index=True, nullable=False)
        ... other fields ...
    )

需要筛选出**至少包含一条private_trade.hedged=False**的order_hist记录，对应的目标SQL逻辑如下：

select
    order_hist.exchange,
    order_hist.order_id,
    count(private_trade.id)
from order_hist 
left join private_trade on private_trade.order_id = order_hist.order_id
left join algo_order on algo_order.id = order_hist.algo_order_id
where 
    private_trade.hedged = False
group by order_hist.exchange, order_hist.order_id

目前用SQLAlchemy实现的代码可以运行，但需要两次查询数据库，不够优雅：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.orm import scoped_session
connstr : str = "postgresql://postgres:xxx@localhost/postgres"
engine = create_engine(connstr)
Session = scoped_session(sessionmaker(bind=engine))
session = Session()

algo_order_ids : Tuple = [ ... ]
statuses : Tuple[str] = (ExchangeOrderStatus.OPEN.name,),
has_at_least_one_unhedged_private_trade : bool = True

orders = session.query(
    Order
).filter(Order.algo_order_id.in_(algo_order_ids)).filter(Order.status.in_(statuses)).order_by(Order.created.asc())

if has_at_least_one_unhedged_private_trade:
    orders_with_unhedged_private_trades = session.query(
        Order.order_id
    ).join(
        PrivateTrade, Order.order_id==PrivateTrade.order_id
    ).filter(Order.algo_order_id.in_(algo_order_ids)).filter(Order.status.in_(statuses)).filter(PrivateTrade.hedged==False).group_by(Order.order_id).having(func.count(PrivateTrade.id)>0).all()
    if len(orders_with_unhedged_private_trades)>0:
        orders_with_unhedged_private_trades = [x[0] for x in orders_with_unhedged_private_trades]
        orders = orders.filter(Order.order_id.in_(orders_with_unhedged_private_trades))
        
in_scope_orders = orders.all()

希望通过SQLAlchemy的.having和func.count方法重构，实现仅单次查询数据库的目标，该如何操作？

可以通过内关联PrivateTrade+分组+having条件直接在主查询中完成筛选，不需要先查order_id列表再过滤。以下是两种可行的重构方式：

方式一：直接分组查询完整Order对象（适合PostgreSQL等支持分组包含非聚合字段的数据库） #

from sqlalchemy import func

# 初始化部分不变
connstr : str = "postgresql://postgres:xxx@localhost/postgres"
engine = create_engine(connstr)
Session = scoped_session(sessionmaker(bind=engine))
session = Session()

algo_order_ids : Tuple = [ ... ]
statuses : Tuple[str] = (ExchangeOrderStatus.OPEN.name,)
has_at_least_one_unhedged_private_trade : bool = True

# 构建主查询
query = session.query(Order)
query = query.filter(Order.algo_order_id.in_(algo_order_ids))
query = query.filter(Order.status.in_(statuses))

if has_at_least_one_unhedged_private_trade:
    # 关联PrivateTrade并过滤未对冲记录
    query = query.join(PrivateTrade, Order.order_id == PrivateTrade.order_id)
    query = query.filter(PrivateTrade.hedged == False)
    # 按Order的所有必要字段分组（需符合数据库分组规则）
    query = query.group_by(Order.id, Order.order_id, Order.algo_order_id, Order.status, Order.created)
    # 确保至少有一条符合条件的交易
    query = query.having(func.count(PrivateTrade.id) > 0)

# 排序并执行单次查询
in_scope_orders = query.order_by(Order.created.asc()).all()

方式二：子查询获取符合条件的order_id，再关联主表（兼容性更强） #

如果数据库不允许分组包含非聚合字段，推荐用子查询先获取符合条件的订单ID，再在主查询中过滤，依然是单次查询：

from sqlalchemy import func

# 初始化部分不变
connstr : str = "postgresql://postgres:xxx@localhost/postgres"
engine = create_engine(connstr)
Session = scoped_session(sessionmaker(bind=engine))
session = Session()

algo_order_ids : Tuple = [ ... ]
statuses : Tuple[str] = (ExchangeOrderStatus.OPEN.name,)
has_at_least_one_unhedged_private_trade : bool = True

# 构建主查询
query = session.query(Order)
query = query.filter(Order.algo_order_id.in_(algo_order_ids))
query = query.filter(Order.status.in_(statuses))

if has_at_least_one_unhedged_private_trade:
    # 子查询：获取有至少一条未对冲交易的order_id
    subquery = session.query(Order.order_id)
    subquery = subquery.join(PrivateTrade, Order.order_id == PrivateTrade.order_id)
    subquery = subquery.filter(Order.algo_order_id.in_(algo_order_ids))
    subquery = subquery.filter(Order.status.in_(statuses))
    subquery = subquery.filter(PrivateTrade.hedged == False)
    subquery = subquery.group_by(Order.order_id)
    subquery = subquery.having(func.count(PrivateTrade.id) > 0).subquery()
    
    # 主查询过滤子查询中的order_id
    query = query.filter(Order.order_id.in_(subquery))

# 排序并执行单次查询
in_scope_orders = query.order_by(Order.created.asc()).all()

说明 #

• 两种方式都只执行一次数据库查询，避免了原代码中两次查询的冗余
• 方式一更直接，但需要根据数据库特性调整分组字段（比如PostgreSQL开启ONLY_FULL_GROUP_BY时需明确列出所有非聚合字段）
• 方式二兼容性更好，所有支持SQL的数据库都能运行

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