Skip to content

Instantly share code, notes, and snippets.

@sonthonaxrk
Last active February 24, 2022 10:52
Show Gist options
  • Save sonthonaxrk/e94dca906b34b7f90648abdc5b704a06 to your computer and use it in GitHub Desktop.
Save sonthonaxrk/e94dca906b34b7f90648abdc5b704a06 to your computer and use it in GitHub Desktop.
Contains eager, multiple of same type
from sqlalchemy import *
from sqlalchemy.orm import *
Base = declarative_base()
class Account(Base):
__tablename__ = "account"
id = Column(Integer, primary_key=True)
name = Column(String)
def __repr__(self):
return f'<Account {self.name}>'
class Customer(Base):
__tablename__ = "customer"
id = Column(Integer, primary_key=True)
account_id = Column(ForeignKey('account.id'))
account = relationship(Account, foreign_keys=[account_id])
name = Column(String)
def __repr__(self):
return f'<Customer {self.name}>'
class Order(Base):
__tablename__ = "user"
id = Column(Integer, primary_key=True)
customer_id = Column(ForeignKey('customer.id'))
account_id = Column(ForeignKey('account.id'))
account = relationship(Account, foreign_keys=[account_id])
customer = relationship(Customer, foreign_keys=[customer_id])
name = Column(String)
def __repr__(self):
return f'<Order {self.name}>'
e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)
account1 = Account(id=1, name="account1")
account2 = Account(id=2, name="account2")
customer = Customer(id=1, name="customer", account_id=1)
order = Order(id=1, name="order", account_id=2)
s = Session(e)
s.add_all([account1, account2, customer, order])
s.commit()
customer_account_alias = aliased(Account, name="customer_account")
def test_query_simple():
base_query = (
select(
Order,
)
.outerjoin(
Account,
Account.id == Order.account_id
)
.outerjoin(
Customer,
Customer.id == Customer.account_id
)
.options(
contains_eager(
Order.customer.of_type(Customer),
).options(
contains_eager(
Customer.account.of_type(Account),
)
)
)
)
res = s.execute(base_query).all()
order = res[0][0]
# This does work
assert order.account_id == order.account.id
# This does not.
assert order.customer.account_id == order.customer.account.id
def test_query_with_single_alias():
"""
First simple attempt. Fails
"""
base_query = (
select(
Order,
)
.outerjoin(
Account,
Account.id == Order.account_id
)
.outerjoin(
Customer,
Customer.id == Customer.account_id
)
.outerjoin(
customer_account_alias,
customer_account_alias.id == Order.customer_id
)
.options(
contains_eager(Order.customer.of_type(Customer)).options(
contains_eager(Customer.account.of_type(Customer))
),
)
)
res = s.execute(base_query).all()
order = res[0][0]
# This does work
assert order.account_id == order.account.id
# This does not.
assert order.customer.account_id == order.customer.account.id
def test_query_with_aliases():
order_account_alias = aliased(Account, name='order_account')
customer_account_alias = aliased(Account, name='customer_account')
base_query = (
select(
Order,
Customer,
order_account_alias,
customer_account_alias,
)
.outerjoin(
order_account_alias,
order_account_alias.id == Order.account_id
)
.outerjoin(
Customer,
Customer.id == Customer.account_id
)
.outerjoin(
customer_account_alias,
customer_account_alias.id == Order.customer_id
)
.options(
contains_eager(
Order.account.of_type(Account),
alias=order_account_alias
),
contains_eager(
Order.customer.of_type(Customer),
).options(
contains_eager(
Customer.account.of_type(Customer),
alias=customer_account_alias
),
),
)
)
res = s.execute(base_query).all()
order = res[0][0]
# This does work
assert order.account_id == order.account.id
# This does not.
assert order.customer.account_id == order.customer.account.id
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment