使用 HAVING 选择 SQLAlchemy 列的计数和值答案

【问题标题】：Select the count and value of a SQLAlchemy column using HAVING使用 HAVING 选择 SQLAlchemy 列的计数和值
【发布时间】：2017-10-28 00:31:31
【问题描述】：

我想选择具有相同电子邮件地址且具有多个重复项的所有联系人的计数。我无法让这个查询在带有 PostgreSQL 的 SQLAlchey 中工作。

SELECT count(*), email FROM contact group by email having count(*) > 1

我试过了：

all_records = db.session.query(Contact).options(
    load_only('email')).group_by(Contact.email).having(
    func.count('*') > 1).all()

sqlalchemy.exc.ProgrammingError: (psycopg2.ProgrammingError) column "contact.id" must appear in the GROUP BY clause or be used in an aggregate function
LINE 1: SELECT contact.id AS contact_id, contact.email AS contact_em...
           ^
[SQL: 'SELECT contact.id AS contact_id, contact.email AS contact_email \nFROM contact GROUP BY contact.email \nHAVING count(%(count_1)s) > %(count_2)s'] [parameters: {'count_1': '*', 'count_2': 1}]

我试过这个：

all_records = db.session.query(func.count(Contact.id)).options(
    load_only('email')).group_by(Contact.email).having(
    func.count('*') > 1).all()

sqlalchemy.exc.ArgumentError
sqlalchemy.exc.ArgumentError: Wildcard loader can only be used with exactly one entity.  Use Load(ent) to specify specific entities.

如果我执行原始 SQL，它可以正常工作：

all_records = db.session.execute(
    "SELECT count(*), email FROM contact group by email"
    " having count(*) > 1").fetchall()

我正在使用 Flask-SQLAlchemy，但这里有一个最小的 SQLAlchemy 设置来演示这个问题：

import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Contact(Base):
    __tablename__ = 'contact'
    id = sa.Column(sa.Integer, primary_key=True)
    email = sa.Column(sa.String)

engine = sa.create_engine('postgresql:///example', echo=True)
Base.metadata.create_all(engine)
session = orm.Session(engine)
session.add_all((
    Contact(email='a@example.com'),
    Contact(email='b@example.com'),
    Contact(email='a@example.com'),
    Contact(email='c@example.com'),
    Contact(email='a@example.com'),
))
session.commit()

# first failed query
all_records = session.query(Contact).options(
    orm.load_only('email')).group_by(Contact.email).having(
    sa.func.count('*') > 1).all()

# second failed query
all_records = db.session.query(sa.func.count(Contact.id)).options(
    orm.load_only('email')).group_by(Contact.email).having(
    sa.func.count('*') > 1).all()

对于样本数据，我希望得到一个结果行，3, a@example.com。

【问题讨论】：

至于为什么指定了load_only()还要添加id，见this。

标签： python postgresql sqlalchemy

【解决方案1】：

您在 SQLAlchemy 中构建的查询与您手动编写的查询不同。

您想选择每封电子邮件出现多次的次数。

q = session.query(
    db.func.count(Contact.email),
    Contact.email
).group_by(
    Contact.email
).having(
    db.func.count(Contact.email) > 1
)
print(q)

SELECT count(contact.email) AS count_1, contact.email AS contact_email 
FROM contact GROUP BY contact.email 
HAVING count(contact.email) > %(count_2)s

第一个查询失败，因为您查询了整个模型，因此 SQLAlchemy 选择了所有列。使用group_by 时只能选择分组列。 SQLAlchemy 在查询整个模型时必须始终选择主键，load_only 不会影响。

第二个查询失败，因为load_only 仅在选择整个模型时有效，但您选择的是聚合和列。

【讨论】：

【解决方案2】：

只需在文本查询中选择您想要的内容：

db.session.query(func.count('*'), Contact.email).\
    group_by(Contact.email).\
    having(func.count('*') > 1).\
    all()

【讨论】：