You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

SQLAlchemy:如何精确匹配多对多关联的作者列表查询项目

问题描述

我正在创建一个包含不同项目的模型,每个项目拥有一位或多位作者。由于每位作者可参与多个项目,这是一种多对多关系,模型设置如下:

from typing import List
from typing import Optional

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy import ForeignKey
from sqlalchemy import ForeignKeyConstraint
from sqlalchemy import Table
from sqlalchemy import Column
from sqlalchemy import UniqueConstraint

class Base(DeclarativeBase):
    pass

author_project_association = Table(
    "author_project_associations",
    Base.metadata,
    Column(
        "author_id",
        ForeignKey("authors.id", onupdate="CASCADE", ondelete="CASCADE"),
        primary_key=True,
    ),
    Column(
        "project_id",
        primary_key=True,
    ),
    Column(
        "project_name",
        primary_key=True,
    ),
    ForeignKeyConstraint(
        ["project_id", "project_name"], ["projects.id", "projects.name"]
    ),
)


class Author(Base):
    __tablename__ = "authors"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(unique=True)

    projects: Mapped[List["Project"]] = relationship(
        back_populates="authors", secondary=author_project_association
    )


class Project(Base):
    __tablename__ = "projects"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(primary_key=True)

    authors: Mapped[List[Author]] = relationship(
        back_populates="projects", secondary=author_project_association
    )
    calculations: Mapped[List["Calculation"]] = relationship(back_populates="project")

目前我根据项目名称和作者列表查询项目的语句如下:

name = "Sample"
authors = ["John", "Kate"]

project = session.scalars(
    select(Project)
    .where(*[Project.authors.any(Author.name == current) for current in authors])
    .where(Project.name == name)
).first()

这段代码能确保返回的项目包含所有指定作者,但可能存在额外未指定的作者。我需要限制返回的项目恰好匹配输入的作者列表,试过在WHERE子句中加入len(Project.authors) == len(authors)这类逻辑但无效,也没能成功嵌入计数子查询。


解决方案

方法一:子查询计数匹配

通过子查询统计项目关联的作者数量,同时确保所有指定作者都存在:

from sqlalchemy import select, func

name = "Sample"
authors = ["John", "Kate"]

# 子查询:统计每个项目的作者数量
author_count_subq = (
    select(
        author_project_association.c.project_id,
        author_project_association.c.project_name,
        func.count(author_project_association.c.author_id).label("author_count")
    )
    .group_by(author_project_association.c.project_id, author_project_association.c.project_name)
    .subquery()
)

project = session.scalars(
    select(Project)
    # 确保包含所有指定作者
    .where(*[Project.authors.any(Author.name == current) for current in authors])
    .where(Project.name == name)
    # 关联子查询,确保作者数量和输入列表长度一致
    .join(
        author_count_subq,
        (Project.id == author_count_subq.c.project_id) & (Project.name == author_count_subq.c.project_name)
    )
    .where(author_count_subq.c.author_count == len(authors))
).first()

方法二:关联后分组过滤

直接关联作者表,分组后通过having子句同时匹配作者数量和指定作者集合:

from sqlalchemy import select, func

name = "Sample"
authors = ["John", "Kate"]

project = session.scalars(
    select(Project)
    .join(Project.authors)
    .where(Project.name == name)
    .where(Author.name.in_(authors))
    .group_by(Project.id, Project.name)
    # 确保分组后的作者数量等于输入列表长度
    .having(func.count(Author.id) == len(authors))
    # 额外确保所有指定作者都被包含(避免重复作者的极端情况)
    .having(func.array_agg(Author.name).contains(authors))
).first()

方法三:EXISTS + NOT EXISTS 精确匹配

通过EXISTS确保所有指定作者都在项目中,同时用NOT EXISTS排除存在其他作者的项目:

from sqlalchemy import exists

name = "Sample"
authors = ["John", "Kate"]

project = session.scalars(
    select(Project)
    .where(Project.name == name)
    # 确保每个指定作者都关联到该项目
    .where(*[
        exists(
            select(1)
            .select_from(author_project_association)
            .join(Author)
            .where(
                (author_project_association.c.project_id == Project.id) &
                (author_project_association.c.project_name == Project.name) &
                (Author.name == current)
            )
        ) for current in authors
    ])
    # 确保项目没有其他未指定的作者
    .where(~exists(
        select(1)
        .select_from(author_project_association)
        .join(Author)
        .where(
            (author_project_association.c.project_id == Project.id) &
            (author_project_association.c.project_name == Project.name) &
            ~Author.name.in_(authors)
        )
    ))
).first()

内容的提问来源于stack exchange,提问作者Raven

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.21 16:05:41