Skip to content

Instantly share code, notes, and snippets.

@vfr292

vfr292/discriminator_on_association_with_eager_loading.py

Created November 25, 2014 07:01
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save vfr292/3330037cf5bc621d3d4b to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save vfr292/3330037cf5bc621d3d4b to your computer and use it in GitHub Desktop.
Download ZIP
discriminator_on_association_with_eager_loading.py
Raw
discriminator_on_association_with_eager_loading.py
'''
Created on Oct 22, 2014
@author: Victor
On 20141124 I asked about how to eager load the parent. The code, I submitted with the post is located at:https://gist.github.com/vfr292/a5939418285e4c8bd03b
'''
"""discriminator_on_related.py
Illustrates a mixin which provides a generic association
using a single target table and a single association table,
referred to by all parent tables. The association table
contains a "discriminator" column which determines what type of
parent object associates to each particular row in the association
table.
SQLAlchemy's single-table-inheritance feature is used
to target different association types.
This configuration attempts to simulate a so-called "generic foreign key"
as closely as possible without actually foregoing the use of real
foreign keys. Unlike table-per-related and table-per-association,
it uses a fixed number of tables to serve any number of potential parent
objects, but is also slightly more complex.
"""
from sqlalchemy import create_engine, Integer, Column, \
String, ForeignKey
from sqlalchemy.orm import Session, relationship, backref, contains_eager
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.ext.declarative import as_declarative, declared_attr
from sqlalchemy.orm import joinedload, with_polymorphic
@as_declarative()
class Base(object):
"""Base class which provides automated table name
and surrogate primary key column.
"""
@declared_attr
def __tablename__(cls):
return cls.__name__.lower()
id = Column(Integer, primary_key=True)
class AddressAssociation(Base):
"""Associates a collection of Address objects
with a particular parent.
"""
__tablename__ = "address_association"
discriminator = Column(String)
"""Refers to the type of parent."""
__mapper_args__ = {"polymorphic_on": discriminator}
class Address(Base):
"""The Address class.
This represents all address records in a
single table.
"""
association_id = Column(Integer, ForeignKey("address_association.id"))
street = Column(String)
city = Column(String)
zip = Column(String)
association = relationship("AddressAssociation", backref="addresses")
parent = association_proxy("association", "parent")
def __repr__(self):
return "%s(street=%r, city=%r, zip=%r)" % \
(self.__class__.__name__, self.street,
self.city, self.zip)
class HasAddresses(object):
"""HasAddresses mixin, creates a relationship to
the address_association table for each parent.
"""
@declared_attr
def address_association_id(cls):
return Column(Integer, ForeignKey("address_association.id"))
@declared_attr
def address_association(cls):
name = cls.__name__
discriminator = name.lower()
cls.assoc_cls = type(
"%sAddressAssociation" % name,
(AddressAssociation, ),
dict(
__tablename__=None,
__mapper_args__={
"polymorphic_identity": discriminator
}
)
)
cls.addresses = association_proxy(
"address_association", "addresses",
creator=lambda addresses: cls.assoc_cls(addresses= addresses)
)
return relationship(cls.assoc_cls,
backref=backref("parent", uselist=False))
class Customer(HasAddresses, Base):
name = Column(String)
sales_person_id = Column(Integer, ForeignKey("salesperson.id"))
def __repr__(self):
return "%s(name = %s)" % (self.__class__.__name__, self.name)
class Supplier(HasAddresses, Base):
company_name = Column(String)
def __repr__(self):
return "%s(name = %s)" % (self.__class__.__name__, self.company_name)
class SalesPerson(Base):
name = Column(String)
customers = relationship('Customer', backref = 'salesperson')
engine = create_engine('sqlite://')
Base.metadata.create_all(engine)
session = Session(engine)
session.add_all([
SalesPerson( name = 'Bob, the would be eager sales person', customers = [
Customer(
name='customer Charlie',
addresses=[
Address(
street='123 anywhere street',
city="New York",
zip="10110"),
Address(
street='40 main street',
city="San Francisco",
zip="95732")
]
),
Customer(
name='customer Lucy',
addresses=[
Address(
street='1 Madison Ave',
city="New York",
zip="11111"),
Address(
street='1 main street',
city="Hometown",
zip="00000")
]
)]),
Supplier(
company_name="Ace Hammers",
addresses=[
Address(
street='2569 west elm',
city="Detroit",
zip="56785")
]
),
])
session.commit()
engine.echo = True
# for customer in session.query(Customer):
# for address in customer.addresses:
# print(address)
# print(address.parent)
print 'START EAGER LOADING TEST'
#ager_sales_persons = session.query(SalesPerson).options(joinedload(SalesPerson.customers).joinedload(Customer.addresses)).all()
#raises AttributeError: 'AssociationProxy' object has no attribute 'property'
# eager_sales_persons = session.query(SalesPerson).options(joinedload(SalesPerson.customers).joinedload(Customer.address_association).joinedload(AddressAssociation.addresses)).all()
#emits SQL for address.parent, but not customer.addresses :)
# eager_sales_persons = session.query(SalesPerson).options(joinedload(SalesPerson.customers).joinedload(Customer.address_association).\
# joinedload(AddressAssociation.addresses).joinedload(Address.association).joinedload(Customer.assoc_cls.parent)).all()
#doesn't join to customer_2 and emits SQL for address.parent
#per 20141124 post by M. Bayer, confirmed address eager loading.
eager_sales_persons = session.query(SalesPerson).options(
joinedload(SalesPerson.customers).
joinedload(Customer.addresses.attr[0]).
joinedload(Customer.addresses.attr[1])
).all()
# for a in Customer.addresses.attr:
# print a
# Customer.address_association
# CustomerAddressAssociation.addresses
print 'START CUSTOMER LOOP'
for eager_sales_person in eager_sales_persons:
for customer in eager_sales_person.customers:
for address in customer.addresses:
print(address)
# print(address.parent)
print 'START ADDRESS EAGER LOADING TEST'
session.expire_all()
#eager_addresses = session.query(Address).options(joinedload(Address.parent))
#error sqlalchemy.exc.ArgumentError: mapper option expects string key or list of attributes
#eager_addresses = session.query(Address).options(joinedload(Address.association).joinedload(Customer.assoc_cls.parent), joinedload(Address.association).joinedload(Supplier.assoc_cls.parent)).all()
#not joining to Customer or Supplier
#eager_addresses = session.query(Address).outerjoin(Address.association).outerjoin(Customer.assoc_cls.parent).options(contains_eager(Address.parent)).outerjoin(Supplier.assoc_cls.parent).options(contains_eager(Address.parent).all())
#error sqlalchemy.exc.ArgumentError: mapper option expects string key or list of attributes
#eager_addresses = session.query(Address).all()
poly = with_polymorphic(
AddressAssociation,
[Customer.assoc_cls, Supplier.assoc_cls], aliased=True)
eager_addresses = session.query(Address).options(
joinedload(Address.association.of_type(poly)).joinedload(
poly.CustomerAddressAssociation.parent),
joinedload(Address.association.of_type(poly)).joinedload(
poly.SupplierAddressAssociation.parent),
).order_by(poly.discriminator.desc(), poly.id)
print 'START ADDRESS LOOP'
for address in eager_addresses:
print address
print address.parent
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment