from sqlalchemy import Column, Text, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.future import select
from sqlalchemy.orm import aliased
from sqlalchemy.sql.visitors import replacement_traverse
Base = declarative_base()
class Foo(Base):
__tablename__ = 'foo'
id = Column(Integer, primary_key=True)
name = Column(Text)
def __repr__(self):
class Bar(Base):
__tablename__ = 'bar'
id = Column(Integer, primary_key=True)
name = Column(Text)
foo_id = Column(Integer)
def __repr__(self):
class Baz(Base):
__tablename__ = 'baz'
id = Column(Integer, primary_key=True)
name = Column(Text)
foo_id = Column(Integer)
def __repr__(self):
baz_alias = select(Baz.__table__).alias()
statement = select(Foo).join(baz_alias, Foo.id == baz_alias.c.foo_id)
def visitor(element):
# print(getattr(element, 'name', None), element is baz_alias)
print('Replacement happened!')
return select(Baz).filter(Baz.name == 'abc').alias()
# print(statement)
print(replacement_traverse(statement, {}, visitor))