如何用Flask SQLAlchemy统计拥有证书的用户总数？

现有数据模型 #

class User(db.Model):
    id = Column(Integer(), primary_key=True)
    name = Column(String(200))
    first_name = Column(String(200))

class Certificates(db.Model):
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("user.id", ondelete="CASCADE"))
    report_sent = Column(Boolean, default=False)
    delivery_time = Column(DateTime)
    certificate_type = Column(String(32))

需求：统计拥有有效证书（certificate_type不为空）的用户总数（注：一个用户可能有多条证书记录，但仅需计数一次）

错误原因分析 #

• 你第一次的查询没有对用户ID去重，每条证书记录都会对应一次User.id的计数，最终返回的是证书总条数而非用户数。
• 第二次的循环查询不仅效率极低（多次发起数据库请求），而且逻辑错误，重复统计了同一用户的多条证书记录。

正确实现方法 #

方法1：关联查询+去重计数

# 直接统计去重后的用户ID数量
user_count = db.session.query(func.count(func.distinct(User.id)))\
    .join(Certificates, User.id == Certificates.user_id)\
    .filter(Certificates.certificate_type.isnot(None))\
    .scalar()

func.distinct(User.id)确保每个用户仅被计数一次，scalar()直接返回数值结果，比all()更简洁。

方法2：子查询筛选用户ID后计数

# 先提取所有拥有有效证书的不重复user_id
subquery = db.session.query(Certificates.user_id)\
    .filter(Certificates.certificate_type.isnot(None))\
    .distinct()\
    .subquery()

# 统计这些ID对应的用户数量
user_count = db.session.query(func.count(User.id))\
    .filter(User.id.in_(subquery))\
    .scalar()

逻辑清晰，适合需要复用子查询结果的场景。

方法3：使用exists子查询

from sqlalchemy import and_

user_count = db.session.query(func.count(User.id))\
    .filter(db.exists().where(
        and_(
            Certificates.user_id == User.id,
            Certificates.certificate_type.isnot(None)
        )
    )).scalar()

通过exists判断用户是否存在有效证书记录，数据量较大时性能更优。

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