Skip to content

Instantly share code, notes, and snippets.

@zzzeek
Created October 7, 2021 01:52
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 1 You must be signed in to fork a gist
  • Save zzzeek/1a612ac70c6a118164e78f13a0f5e42f to your computer and use it in GitHub Desktop.
Save zzzeek/1a612ac70c6a118164e78f13a0f5e42f to your computer and use it in GitHub Desktop.
adapt_on_lookup_mapping demo
from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import inspect
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import Table
from sqlalchemy.orm import aliased
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import Session
Base = declarative_base()
class A(Base):
__tablename__ = "a"
id = Column(Integer, primary_key=True)
data = Column(String)
not_a_at_all = Table("a", MetaData(), Column("id"), Column("data"))
e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)
s = Session(e)
s.add(A(data="foo"))
s.add(A(data="bar"))
s.commit()
# "not_a_at_all" is a totally unrelated Table object with matching col names
na1 = not_a_at_all.alias("a")
na2 = not_a_at_all.alias("b")
# full statement w join etc.
stmt = select(na1, na2).outerjoin_from(na1, na2, na1.c.id > na2.c.id)
# subquery we will select ORM entities from
subq = stmt.subquery()
aa = aliased(A, subq)
bb = aliased(A, subq)
# there is private state in the aliased() object that is essentially a
# "lookup table". it is theoreticaly possible to create a public API that
# uses this system
inspect(aa)._adapter.equivalents = {
A.__table__.c.id: [na1.c.id],
A.__table__.c.data: [na1.c.data],
}
inspect(bb)._adapter.equivalents = {
A.__table__.c.id: [na2.c.id],
A.__table__.c.data: [na2.c.data],
}
print(s.execute(select(aa, bb)).all())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment