SQLAlchemy 2.0基于关联关系的事件按来源+国家聚合查询实现

1. 调整模型配置 #

首先补充模型的关联关系，以及可选的查询字段声明，消除后续属性访问的IDE警告：

import uuid
from sqlalchemy import Column, String, UUID, ForeignKey, DateTime, func
from sqlalchemy.orm import declarative_base, relationship, query_expression

Base = declarative_base()

class Country(Base):
    __tablename__ = "countries"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, index=True)
    code = Column(String, index=True, nullable=False, unique=True)
    name = Column(String, nullable=False)
    # 声明查询时注入的统计字段，可选，用于消除IDE属性不存在警告
    events_count = query_expression()
    # 反向关联事件，可选
    events = relationship("Event", back_populates="country", cascade="all, delete-orphan")


class EventSource(Base):
    __tablename__ = "eventsources"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, index=True)
    name = Column(String, nullable=False, unique=True)
    # 反向关联事件，可选
    events = relationship("Event", back_populates="eventsource", cascade="all, delete-orphan")
    # 多对多关联到国家，中间表为events，只读用于统计
    countries = relationship(
        "Country",
        secondary="events",
        viewonly=True
    )


class Event(Base):
    __tablename__ = "events"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, index=True)
    country_id = Column(
        UUID(as_uuid=True),
        ForeignKey("countries.id", ondelete="CASCADE"),
        nullable=False,
    )
    eventsource_id = Column(
        UUID(as_uuid=True),
        ForeignKey("eventsources.id", ondelete="CASCADE"),
        nullable=False,
    )
    created_at = Column(DateTime(timezone=True), default=func.now())
    # 关联所属国家和事件源
    country = relationship("Country", back_populates="events")
    eventsource = relationship("EventSource", back_populates="events")

2. 编写SQLAlchemy 2.0风格查询 #

用SQLAlchemy 2.0的contains_eager和with_expression语法，直接在查询中注入聚合统计字段，无需额外子查询：

from sqlalchemy import select, func
from sqlalchemy.orm import contains_eager

# 构造查询语句
stmt = (
    select(EventSource)
    # 关联事件表和国家表
    .join(EventSource.events)
    .join(Event.country)
    # 按事件源、国家分组统计事件数
    .group_by(EventSource.id, Country.id)
    .options(
        # 声明当前查询关联的Country实例就是EventSource.countries的关联数据，无需额外加载
        contains_eager(EventSource.countries)
        # 给Country实例动态注入events_count字段，值为聚合统计的事件数量
        .with_expression(Country.events_count, func.count(Event.id))
    )
    # 按事件源名称排序
    .order_by(EventSource.name)
)

# 执行查询，用unique方法去重重复的EventSource实例
eventsources = session.execute(stmt).unique().scalars().all()

效果验证 #

你提供的测试数据执行后，调用要求的输出语句：

print(eventsources[0].countries[0].events_count)

会正确输出2，完全符合你的需求。 #

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