I have a schema that includes, among other things, self-joins and many-to-many relationships, like this:
from typing import List, Optional
from sqlalchemy import create_engine
from sqlalchemy.orm import (
aliased,
DeclarativeBase,
Session,
Mapped,
mapped_column,
relationship,
)
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.schema import ForeignKey
from sqlalchemy.types import String
class Base(DeclarativeBase):
pass
class Post(Base):
__tablename__ = 'post'
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(200), nullable=False)
parent_id: Mapped[int] = mapped_column(ForeignKey('post.id'), nullable=True)
parent: Mapped["Post"] = relationship('Post', foreign_keys=parent_id, remote_side=id)
tags: Mapped[List['Tag']] = relationship('Tag', secondary='tag2post')
class Tag(Base):
__tablename__ = 'tag'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100), nullable=False)
class Tag2Post(Base):
__tablename__ = 'tag2post'
id: Mapped[int] = mapped_column(primary_key=True)
tag_id: Mapped[int] = mapped_column('tag_id', ForeignKey('tag.id'))
tag: Mapped[Tag] = relationship(Tag, overlaps='tags')
post_id: Mapped[int] = mapped_column('post_id', ForeignKey('post.id'))
post: Mapped[Post] = relationship(Post, overlaps='tags')
engine = create_engine("sqlite+pysqlite:///:memory:", echo=True)
Base.metadata.create_all(engine)
with Session(engine) as session:
session.add(tag_a := Tag(name='a'))
session.add(tag_b := Tag(name='b'))
session.add(parent := Post(title='parent', tags=[tag_a]))
session.add(child := Post(title='child', parent=parent, tags=[tag_b]))
Now I want to write a query of the form “give me all posts with tag a
and whose parent has a tag b
“.
I understand the “manual” way of adding aliases for each step:
parent_alias = aliased(Post)
parent_tag = aliased(Tag)
parent_tag2post = aliased(Tag2Post)
q = session.query(
Post
).join(
parent_alias, Post.parent_id == parent_alias.id
).join(
parent_tag2post,
parent_alias.id == parent_tag2post.post_id
).join(
parent_tag,
parent_tag2post.tag_id == parent_tag.id,
).join(
Post.tags,
).filter(
parent_tag.name == 'a',
Tag.name == 'b',
)
print(q.one().title) # prints 'child'
But I’m dealing with more generic code, basically a query language that gives me the conditions:
Post.tags.name == 'b'
Post.parent.tags.name == 'a'
The code that translates the query language into JOINs now would need to introspect the relationships (like Post.tags
, Post.parents
, Post.parents.tags
), and in each relationship replace the tables by aliases and rebuild the join conditions with the aliases… which sounds rather complicated and error-prone.
Is there an easier way? Like, maybe, telling sqlalchemy to use a relationship, but with an alias? Or is there maybe a third-party package that already does this?