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:

1. User creates a set (which inserts new row to sets)
2. 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, ...))

1. if SetEntity.ownerId == userId - OK
2. 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};

Answer 1

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