我使用 Flask-SQLAlchemy 定义了以下模型:
"""models.py"""
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
skill_candidate = db.Table(
'SkillCandidate',
db.Column('skill_id', db.String, db.ForeignKey('skill.id')),
db.Column('candidate_id', db.Integer, db.ForeignKey('candidate.id')))
class Candidate(db.Model):
id = db.Column(db.Integer, primary_key=True)
skills = db.relationship("Skill", secondary=skill_candidate)
class Skill(db.Model):
id = db.Column(db.String, primary_key=True)
name = db.Column(db.String, nullable=False, unique=True)
我想要实现的目标如下: 我想返回拥有列表输入中提供的技能的所有候选人(甚至理想情况下,是技能 ID 列表)
我尝试了以下方法:
def get_skilled_candidates(skill_ids):
return Candidate.query.join(skill_candidate).\
filter(and_(*[skill_candidate.c.skill_id == skill_id for skill_id in skill_ids])).\
all()
目的是过滤所有候选人的每项技能,并用 and_ 语句组成
如果我使用包含 1 项的列表(它返回拥有该技能的所有候选人),效果很好,但如果我在输入列表中添加更多技能,则效果不佳(即使我的基础中有符合条件的候选人)
要小心这个答案,越短并不总是越好。 @IljaEverilä 使用关系除法的 answer 在许多情况下可能会表现得更好。
您可以查询具有列表中任何技能的所有候选人,然后使用列表理解过滤结果。在许多情况下,这不会像关系划分方法那样高效,但它确实简化了查询方面。
skill_ids = ['id_1', 'id_2']
candidates = session.query(Candidate).\
filter(Candidate.skills.any(Skill.id.in_(skill_ids)).\
all()
candidates = [
c for c in candidates
if set(s.id for s in c.skills).issuperset(skill_ids)
]
如评论中所述,您需要的是
FORALL
运算(通用量词)或关系除法。
FORALL x ( p(x) )
可以表示为
NOT ( EXISTS x ( NOT ( p(x) ) ) )
如果您不了解
FORALL
及其关系,这有点笨拙且难以推理。鉴于您的模型,它可能看起来像:
def get_skilled_candidates(skill_ids):
# Form a temporary derived table using unions
skills = db.union_all(*[
db.select([db.literal(sid).label('skill_id')])
for sid in skill_ids]).alias()
return Candidate.query.\
filter(
~db.exists().select_from(skills).where(
~db.exists().
where(db.and_(skill_candidate.c.skill_id == skills.c.skill_id,
skill_candidate.c.candidate_id == Candidate.id)).
correlate_except(skill_candidate))).\
all()
当然还有其他方式来表达相同的查询,例如:
def get_skilled_candidates(skill_ids):
return Candidate.query.\
join(skill_candidate).\
filter(skill_candidate.c.skill_id.in_(skill_ids)).\
group_by(Candidate.id).\
having(db.func.count(skill_candidate.c.skill_id.distinct()) ==
len(set(skill_ids))).\
all()
本质上是通过计数检查所有技能 ID 是否匹配。
如果使用 Postgresql,你也可以这样做:
from sqlalchemy.dialects.postgresql import array_agg
def get_skilled_candidates(skill_ids):
# The double filtering may seem redundant, but the WHERE ... IN allows
# the query to use indexes, while the HAVING ... @> does the final filtering.
return Candidate.query.\
join(skill_candidate).\
filter(skill_candidate.c.skill_id.in_(skill_ids)).\
group_by(Candidate.id).\
having(array_agg(skill_candidate.c.skill_id).contains(skill_ids)).\
all()
这在某种程度上与其他答案中的部分 Python 解决方案等效。
此外,可以使用聚合
EVERY
:
def get_skilled_candidates(skill_ids):
# Form a temporary derived table using unions
skills = db.union_all(*[
db.select([db.literal(sid).label('skill_id')])
for sid in skill_ids]).alias()
# Perform a CROSS JOIN between candidate and skills
return Candidate.query.\
join(skills, db.true()).\
group_by(Candidate.id).\
having(db.func.every(
db.exists().
where(db.and_(skill_candidate.c.skill_id == skills.c.skill_id,
skill_candidate.c.candidate_id == Candidate.id)).
correlate_except(skill_candidate))).\
all()