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
相关产品推荐
相关产品推荐

