Select ORM objects whose child attributes exactly match a given list

user_19522518 0
Viewed 45

I'm new with SQLAlchemy and I'm facing a problem that I couldn't solve.

Problem: I need to filter people who have exactly and only the skills I'm looking for.

I'm working with one to many relationship. Let's say I have a Person class (Parent) and a Skill class (child) defined as follows:

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")

I have some data in these tables for testing; Represented below:

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

Then the people are listed as below:

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

When I filter by skill "Excel", I want it to return only the person who has only excel as a skill, but when I run the query below:

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

the result is:

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

But the desired result was:

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

Thanks for any kind of help!! Maybe I modeled the tables wrongly; :(

fastapi sqlalchemy python
2 Answers

Nicely asked question.

SQL construct EXISTS comes to my mind. It basically requires 2 table queries (in one command). The idea is to:

  1. Get all people that know Excel (this part you have done)

  2. Get all people that know anything else than Excel

  3. Keep only those people from (1) that don't exist in (2). This is where you can use the NOT EXISTS (in sqlalchemy it's ~) operator (example).

Because your desired people have ONLY the excel skill, you can get them by this set difference operation.

user_87099280

We can break your task down into three steps:

  1. Identify persons with one or more of the desired skills. In other words, eliminate users who don't have any of the desired skills.
  2. Narrow that group to persons who have all of the desired skills.
  3. Eliminate persons having extra (unwanted) skills.

We can to that with a series of queries where each query (step) builds on the previous query.

# 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')]
user_21443900
Related