How do I query properties of elements of a list type attribute of a flask-sqlalchemy model?

user_1585375 0
Viewed 49

I am trying out flask-sqlalchemy and have these two models where one person can have multiple addresses and addresses are strictly associated to one person:

class Person(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), nullable=False)
    addresses = db.relationship('Address', backref='person', lazy=True)

class Address(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(120), nullable=False)
    phone = db.Column(db.String(20), nullable=False)
    person_id = db.Column(db.Integer, db.ForeignKey('person.id'),
        nullable=False)

I want to query the Person table to get a list of persons who have at least one email address in their email address list matching some string. So far I have been able to come up with this:

kwargs = {...} # key-val filter that works fine standalone on Person non address attributes
email_cond = Person.addresses.any(Address.email == substring)
result = Person.query.filter_by(**kwargs).filter(email_cond).limit(10).all()

But this returns one Person object if there is exact match on the email. Also, I'm not sure if this will match more than one. How should I go about this if I want partial match on the attributes of Address (like, partial match on email or phone number)?

Note: I have larger models and more than one attributes as relationships, using this as an example.

flask-sqlalchemy sqlalchemy sql
1 Answers

This seems like a simple thing but there are a lot of options. Using any() can work for this case but might have poor performance if there are a lot of emails or a lot of persons, etc. I tried to show 3 examples below:

  1. using a subquery (I think this is the most flexible)
  2. using distinct with a join
  3. using a correlated subquery with any().

It is helpful to print the queries out and check the SQL being produced to make sure the combination makes sense.


from sqlalchemy import (
    create_engine,
    Integer,
    String,
)
from sqlalchemy.schema import (
    Column,
    MetaData,
    ForeignKey,
)
from sqlalchemy.sql import select
from sqlalchemy.orm import declarative_base, relationship, Session


metadata = MetaData()

Base = declarative_base(metadata=metadata)

engine = create_engine('postgresql+psycopg2://username:password@/database', echo=False)


class Person(Base):
    __tablename__ = "persons"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), nullable=False)
    addresses = relationship('Address', backref='person', lazy=True)



class Address(Base):
    __tablename__ = "addresses"
    id = Column(Integer, primary_key=True)
    person_id = Column(Integer, ForeignKey('persons.id'), nullable=True)
    email = Column(String(120), nullable=False)


metadata.create_all(engine)


with Session(engine) as session:
    person_dicts = [{
        "name": 'dog',
        "emails": ['dog@example.com', 'dog@example1.com', 'dog@example2.com'],
    }, {
        "name": 'cat',
        "emails": ['cat@example1.com', 'cat@example2.com'],
    }, {
        "name": 'bird',
        "emails": ['bird@example.com']
    }]
    for person_dict in person_dicts:
        person = Person(name=person_dict['name'])
        session.add(person)
        for email in person_dict['emails']:
            address = Address(email=email)
            person.addresses.append(address)
            session.add(address)
    session.commit()

    emails_to_match = ['dog@example.com', 'dog@example1.com', 'cat@example2.com']

    # subquery, can add more subqueries to condition or other kwargs
    person_id_subquery = select(Person.id).join(Person.addresses).filter(Address.email.in_(emails_to_match)).scalar_subquery()
    q = session.query(Person).filter(Person.id.in_(person_id_subquery))
    print (q)
    results = q.all()
    for result in results:
        print (result.name)

    # distinct
    q = session.query(Person).join(Person.addresses).filter(Address.email.in_(emails_to_match)).distinct()
    print (q)
    results = q.all()

    for result in results:
        print (result.name)

    # correlated subquery
    q = session.query(Person).filter(Person.addresses.any(Address.email.in_(emails_to_match)))
    print (q)
    results = q.all()
    for result in results:
        print (result.name)

Output

SELECT persons.id AS persons_id, persons.name AS persons_name 
FROM persons 
WHERE persons.id IN (SELECT persons.id 
FROM persons JOIN addresses ON persons.id = addresses.person_id 
WHERE addresses.email IN (__[POSTCOMPILE_email_1]))
dog
cat
SELECT DISTINCT persons.id AS persons_id, persons.name AS persons_name 
FROM persons JOIN addresses ON persons.id = addresses.person_id 
WHERE addresses.email IN (__[POSTCOMPILE_email_1])
dog
cat
SELECT persons.id AS persons_id, persons.name AS persons_name 
FROM persons 
WHERE EXISTS (SELECT 1 
FROM addresses 
WHERE persons.id = addresses.person_id AND addresses.email IN (__[POSTCOMPILE_email_1]))
dog
cat
user_20360700
Related