Hi,
I have this case where I want to be able to access a hybrid_property that
points to a related model which is further filtered using a cte (utilizing
a Transformer
<http://docs.sqlalchemy.org/en/latest/orm/extensions/hybrid.html#building-transformers>)
without having to trigger additional database requests. So what I'm trying
to achieve looks something like this:
# Python code
customers = session.query(Customer).with_transformation(Customer.
latest_sku_shipment.cte).all()
# Jinja2 code
{% for customer in customers %}
<h1>{{ customer.name }}</h1>
<p>Latest shipment, {{ customer.latest_sku_shipment.shipment_date }}, was
sent to customer on {{ customer.latest_sku_shipment.apa }}</p>
{% endfor %}
Here's an example I've built that shows that it tries to access the regular
property. Is there any way that it can get the data from the query that has
already been executed?
import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext import hybrid
Base = declarative_base()
Session = sa.orm.sessionmaker()
class CustomerLatestShipmentTransformer(hybrid.Comparator):
@property
def cte(self):
def transform(q):
subq = (
q.session.query(
ProductSKUShipment.customer_id,
ProductSKUShipment.shipment_date,
ProductSKUShipment.order_no,
)
.distinct(ProductSKUShipment.customer_id) # Postgres
specific
.order_by(
ProductSKUShipment.customer_id,
ProductSKUShipment.shipment_date.desc(),
)
)
cte = subq.cte()
return q.outerjoin(cte, cte.c.customer_id == Customer.id)
return transform
class Customer(Base):
__tablename__ = 'customer'
id = sa.Column(sa.Integer, primary_key=True)
no = sa.Column(sa.Text, nullable=False)
name = sa.Column(sa.Text, nullable=False)
def __repr__(self):
return '<{no}: {name}>'.format(**vars(self))
@hybrid.hybrid_property
def latest_sku_shipment(self):
# sess = sa.orm.object_session(self)
# return (
# sess.query(ProductSKUShipment)
# .filter_by(customer_id=self.id)
# .order_by(ProductSKUShipment.shipment_date.desc())
# .first()
# )
raise RuntimeError('I want to be loaded from my query!')
@latest_sku_shipment.comparator
def latest_sku_shipment(cls):
return CustomerLatestShipmentTransformer(cls)
class ProductSKUShipment(Base):
__tablename__ = 'productskushipment'
def __repr__(self):
return '<{sku} {order_no} ({shipment_date})>'.format(**vars(self))
id = sa.Column(sa.Integer, primary_key=True)
sku = sa.Column(sa.Text, unique=True)
quantity = sa.Column(sa.Integer, nullable=False)
order_no = sa.Column(sa.Text, nullable=False)
shipment_date = sa.Column(sa.DateTime(timezone=True), nullable=False)
customer_id = sa.Column(
sa.Integer,
sa.ForeignKey(Customer.id, ondelete='CASCADE'),
nullable=False)
customer = sa.orm.relationship(
Customer,
backref=sa.orm.backref('sku_shipments', passive_deletes='all'),
)
customer = Customer(
no='123',
name='My customer',
)
sku_shipment = ProductSKUShipment(
customer=customer,
order_no='12345',
sku='20090-100',
quantity=1,
shipment_date='2015-01-01 08:34',
)
sku_shipment_2 = ProductSKUShipment(
customer=customer,
order_no='12345',
sku='10570-900',
quantity=1,
shipment_date='2016-02-19 13:03',
)
engine = sa.create_engine('postgresql://localhost/cachedproptest')
Session.configure(bind=engine)
Base.metadata.create_all(bind=engine)
session = Session()
session.add(customer)
session.flush()
q = (
session.query(Customer)
.with_transformation(Customer.latest_sku_shipment.cte)
# .options(sa.orm.joinedload('latest_sku_shipment'))
)
for customer in q:
print(customer.latest_sku_shipment)
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to sqlalchemy+unsubscr...@googlegroups.com.
To post to this group, send email to sqlalchemy@googlegroups.com.
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
