python - SQLAlchemy join & filter - Stack Overflow

admin管理员组

文章数量:1302374

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...

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};

• python
• join
• filter
• sqlalchemy
• many-to-many

Share Follow edited Feb 10 at 17:03 snakecharmerb 55.9k1313 gold badges132132 silver badges186186 bronze badges asked Feb 10 at 14:06 Emma GrinbergEmma Grinberg 1333 bronze badges 4

• Can you explain your conditions a bit in detail? – python_user Commented Feb 10 at 15:42
• Can your desired query be described as "Return all the Sets that are owned by a specified User."? – 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 Commented Feb 10 at 16:54
• I updated my question with SQL query. – Emma Grinberg Commented Feb 10 at 17:00

Add a comment  | 

1 Answer 1

Sorted by: Reset to default 0

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

本文标签： pythonSQLAlchemy join amp filterStack Overflow