如何在SQLAlchemy中编写数据库无关查询：筛选仅对应Alice的stock

需求说明 #

给定如下SQLAlchemy表模型：

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Data(Base):
    id = Column(Integer, primary_key=True)
    stock = Column(String)
    customer = Column(String)

需要查询所有仅被客户"Alice"关联的stock去重值——即某个stock的所有关联记录中，customer字段只有"Alice"，没有其他值。

示例数据表：

预期结果：["B"]

原实现的兼容性问题 #

此前基于PostgreSQL的实现使用了专属的DISTINCT ON语法，在SQLite、MySQL等数据库中会触发警告甚至报错：

result = session.query(Data.stock).distinct(Data.stock).filter(
    Data.customer == "Alice"
).group_by(Data.stock).having(
    ~Data.stock.in_(session.query(Data.stock).filter(Data.customer != "Alice"))
)

SADeprecationWarning: DISTINCT ON is currently supported only by the PostgreSQL dialect. Use of DISTINCT ON for other backends is currently silently ignored, however this usage is deprecated, and will raise CompileError in a future release for all backends that do not support this syntax.

跨数据库兼容的解决方案 #

方案一：子查询排除法 #

通过子查询找出所有被非"Alice"客户关联的stock，再从关联过"Alice"的stock中排除这些值，最后通过group_by去重：

from sqlalchemy import create_engine, sessionmaker

engine = create_engine('connection_string')
Session = sessionmaker(bind=engine)
session = Session()

# 子查询：获取所有被非Alice客户使用的stock
other_stocks = session.query(Data.stock).filter(Data.customer != "Alice").distinct()

# 主查询：筛选关联过Alice且未被其他客户使用的stock
result = session.query(Data.stock)\
    .filter(Data.customer == "Alice")\
    .filter(~Data.stock.in_(other_stocks))\
    .group_by(Data.stock)

# 提取结果
print([row[0] for row in result])

方案二：分组统计法（更简洁高效） #

通过分组后统计每个stock对应的不同客户数量，同时验证唯一客户为"Alice"，无需子查询：

from sqlalchemy import create_engine, sessionmaker, func

engine = create_engine('connection_string')
Session = sessionmaker(bind=engine)
session = Session()

result = session.query(Data.stock)\
    .group_by(Data.stock)\
    .having(
        # 该stock仅关联1种客户
        func.count(func.distinct(Data.customer)) == 1,
        # 唯一的客户是Alice
        func.max(Data.customer) == "Alice"
    )

# 提取结果
print([row[0] for row in result])

两种方案均兼容MySQL、SQLite、PostgreSQL等主流数据库，且能正确返回预期结果。

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