选择子属性与给定列表完全匹配的ORM对象

Question

我是SQLAlchemy的新手，我面临着一个我无法解决的问题。

问题:我需要筛选那些拥有我正在寻找的技能的人。

我正在处理一对多的关系。假设我有一个Person类(父类)和一个Skill类(子类)，定义如下:

class Person(Base):
__tablename__ = "person"
id = Column(Integer, primary_key=True)
name = Column(String(50))
skills = relationship("Skill", back_populates="person")

class Skill(Base):
__tablename__ = "skill"
id = Column(Integer, primary_key=True)
skill_name = Column(String(20))
person_id = Column(Integer, ForeignKey("person.id"))
person = relationship("Person", back_populates="skills")

我在这些表中有一些数据用于测试;代表如下:

Person (Table)
id=1, name=Kelly
id=2, name=William
id=3, name=Jerry
Skill (Table)
id=1, name=Excel, person_id=1
id=2, name=Excel, person_id=2
id=3, name=Python, person_id=2
id=4, name=Social, person_id=3

那么这些人的名单如下:

id=1, name=Kelly, skills=[Skill(id=1)] # Kelly knows Excel
id=2, name=William, skills=[Skill(id=2), Skill(id=3)] # William knows Excel and Python
id=3, name=Jerry, skills=[Skill(id=4)] # Jerry has social skill

当我按技能"excel"过滤时，我希望它只返回只有excel作为技能的人，但是当我运行下面的查询时:

q = session.query(Person).join(Skill).filter(Skill.name == "Excel").all()

的结果是:

id=1, name=Kelly, skills=[Skill(id=1)] # Kelly knows Excel
id=2, name=William, skills=[Skill(id=2), Skill(id=3)] # William knows Excel and Python

但是期望的结果是:

id=1, name=Kelly, skills=[Skill(id=1)] # Kelly knows Excel

谢谢你的帮助!!也许我建模错误的表;(

Answer 1

问得好。

SQL结构EXISTS出现在我的脑海中。它基本上需要2个表查询(在一个命令中)。这个想法是:

1. 找到所有知道Excel的人(这部分你已经完成了)

2. 让所有知道的人都知道除了Excel

3. 只保留那些人从(1),(2)不存在。在这里您可以使用NOT EXISTS(在sqlalchemy~)操作符(例子)。

因为你想要的人只有excel的技能,你可以通过这一套不同的操作。

Answer 2

我们可以把你的任务分成三步:

1. 识别具有一种或多种所需技能的人员。换句话说，排除那些不具备任何所需技能的用户。
2. 将这一群体缩小到具有所有所需技能的人。
3. 取消拥有额外(不需要的)技能的人员。

我们可以用一系列的查询,其中每个查询(步骤)建立在之前的查询。

# fmt: off
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, select, func, text
from sqlalchemy.orm import declarative_base, relationship, Session
# fmt: on
engine = create_engine("sqlite://")
Base = declarative_base()

class Person(Base):
__tablename__ = "person"
id = Column(Integer, primary_key=True)
name = Column(String(50))
skills = relationship("Skill", back_populates="person")
def __repr__(self):
return f"Person({repr(self.name)})"

class Skill(Base):
__tablename__ = "skill"
id = Column(Integer, primary_key=True)
skill_name = Column(String(20))
person_id = Column(Integer, ForeignKey("person.id"))
person = relationship("Person", back_populates="skills")
def __init__(self, skill_name):
self.skill_name = skill_name

Base.metadata.create_all(engine)
with Session(engine) as sess:
# example data
sess.add_all(
[
Person(name="Kelly", skills=[Skill("Excel")]),
Person(
name="William",
skills=[Skill("Excel"), Skill("Python")],
),
Person(name="Jerry", skills=[Skill("Social")]),
Person(
name="Keener",
skills=[Skill("Excel"), Skill("Python"), Skill("Social")],
),
]
)
sess.commit()
with Session(engine) as sess:
skills_to_match = [
"Excel",
"Python",
]
# Step 1: Select Skill records matching any skill in the list
with_any_skills = select(Skill.person_id, Skill.skill_name).where(
Skill.skill_name.in_(skills_to_match)
)
# Step 2: Use group_by() to identify persons matching all the skills in the list
with_all_skills = (
select(with_any_skills.c.person_id)
.group_by(with_any_skills.c.person_id)
.having(func.count(text("*")) == len(skills_to_match))
.subquery()
)
# Step 3: Eliminate persons having extra (unwanted) skills.
selected_users_total_skill_count = (
select(Skill.person_id, func.count(text("*")).label("num_skills"))
.join(with_all_skills, with_all_skills.c.person_id == Skill.person_id)
.group_by(Skill.person_id)
.subquery()
)
with_only_skills = (
select(with_all_skills.c.person_id)
.join(
selected_users_total_skill_count,
selected_users_total_skill_count.c.person_id
== with_all_skills.c.person_id,
)
.where(
selected_users_total_skill_count.c.num_skills
== len(skills_to_match)
)
)
final_query = select(Person).where(Person.id.in_(with_only_skills))
result = sess.scalars(final_query).all()
print(result)  # [Person('William')]