SQLAlchemy join & filter

Question

I am new to SQLAlchemy and develop a small FastAPI application using SQLAlchemy and SQLite. I have several tables, it's just a small example:

USERS:

class UserEntity(BaseEntity):
    """
    User's table
    """
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, autoincrement=True, unique=True, nullable=False)
    username = Column(String, unique=True, index=True, nullable=False)
    password = Column(String, nullable=False)

SETS:

class SetEntity(BaseEntity):
    """
    Sets
    """
    __tablename__ = "sets"

    id = Column(Integer, primary_key=True, autoincrement=True, unique=True, nullable=False)
    name = Column(String, nullable=False)
    ownerId = Column("owner_id", Integer, ForeignKey("users.id"), nullable=False, comment="Creator's id")

    owner = relationship("UserEntity", backref="sets")
    members = relationship("UserEntity", secondary=SetMemberEntity)

And many-to-many table:

SetMemberEntity = Table(
    "set_members",
    BaseEntity.metadata,
    Column("id", Integer, primary_key=True),
    Column("set_id", Integer, ForeignKey("sets.id")),
    Column("member_id", Integer, ForeignKey("users.id"))
)

So user's process is:

User creates a set (which inserts new row to sets)
User adds another users (by id) who have access to the set created in step 1.

For example:

USERS:

id	username	password
1	Alice	3c9...
2	Bob	bb4...
3	Trent	304...

SETS:

id	name	owner_id
1	Alice's set	1
2	Bob's set	2

SETS_MEMBERS:

id	set_id	member_id
1	1	2
1	1	3
1	2	3

I need to write query filter which returns sets, corresponding conditions:

def getAvailableSets(session: Session, userId: int) -> list[SetEntity]:
    return session.query(SetEntity).join(...).filter(_or(SetEntity.ownerId == userId, ...))

if SetEntity.ownerId == userId - OK
if userId in list(map(lambda member: member.id, SetEntity.members)) - I don't know how to solve it and write in filter function :(

For example:

getAvailableSets(session, 1) -> [SetEntity(id=1, ...)]
getAvailableSets(session, 3) -> [SetEntity(id=1, ...), SetEntity(id=2, ...)]

Is there any option to do this?

UPD. Something like that in SQL:

select s.* from sets s
left join set_members sm on s.id = sm.set_id
where s.owner_id = {user_id} or sm.member_id = {user_id};

Can your desired query be described as "Return all the Sets that are owned by a specified User."? — Gord Thompson
– Gord Thompson, Commented Feb 10 at 16:16
@python_user @gord-thompson I want getAvailableSets method to return sets that are owned by specific user (userId) OR sets user has access to. For example, user Trent owns no sets, but he is member of sets with id 1 and 2. So method will return sets with id 1 and 2. If Trent creates new set with id 3, method will return sets with id 1, 2, 3. — Emma Grinberg
– Emma Grinberg, Commented Feb 10 at 16:54

Gord Thompson · Accepted Answer · 2025-02-10 18:35:10Z

Here is one way to do it:

def getAvailableSets(session: Session, userId: int) -> list[SetEntity]:
    set_ids = (
        select(SetEntity.id)
        .where(SetEntity.ownerId == userId)
        .union(
            select(SetMemberEntity.c.set_id).where(
                SetMemberEntity.c.member_id == userId
            )
        )
    )
    return session.scalars(
        select(SetEntity).where(SetEntity.id.in_(set_ids))
    ).all()


with Session(engine) as sess:
    returned_sets = getAvailableSets(sess, 3)
    """
SELECT sets.id, sets.name, sets.owner_id 
FROM sets 
WHERE sets.id IN (SELECT sets.id 
FROM sets 
WHERE sets.owner_id = ? UNION SELECT set_members.set_id 
FROM set_members 
WHERE set_members.member_id = ?)
2025-02-10 11:16:11,677 INFO sqlalchemy.engine.Engine [generated in 0.00064s] (3, 3)
    """
    print(f">>> {returned_sets}")
    """
>>> [SetEntity(id=1), SetEntity(id=2)]
    """

Collectives™ on Stack Overflow

SQLAlchemy join & filter

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related