-
-
Notifications
You must be signed in to change notification settings - Fork 1.5k
ExpiryMemoized
The SQLAlchemy docs include guidelines for building Python @property objects that query for related data. The advantage to @property is that you can write a straightforward Query in the method body to get exactly the data you want. But one disadvantage to this, compared to column_property() or relationship() is that it evaluates each time it's called. A so-called "memoized" property, of which there are many recipes on the web, is an easy solution here, but then the value can become stale when data changes.
So here we introduce a simple recipe, building on the highest performing "memoized" property (one which replaces itself) to add an event listener that will expire all loaded versions any time the object's parent session is flushed:
#!python
from sqlalchemy.orm import Session
from sqlalchemy import event
import collections
class orm_memoized(object):
"""A read-only @property that is only evaluated once.
Adds a listener so that values are expired
when the object is part of a flush.
"""
all_memoized = collections.defaultdict(set)
def __init__(self, fget, doc=None):
self.fget = fget
self.__doc__ = doc or fget.__doc__
self.__name__ = fget.__name__
def __get__(self, obj, cls):
if obj is None:
return self
self.all_memoized[cls].add(self.__name__)
obj.__dict__[self.__name__] = result = self.fget(obj)
return result
@event.listens_for(Session, "after_flush")
def expire(session, flush_context):
"""expire all orm_memoized en masse for a given flush"""
for obj in session.identity_map.values():
for name in orm_memoized.all_memoized[obj.__class__]:
obj.__dict__.pop(name, None)
A full example is as follows:
#!python
from sqlalchemy import Integer, Column, event, func, ForeignKey, create_engine
from sqlalchemy.orm import relationship, Session, object_session
from sqlalchemy.ext.declarative import declarative_base
import collections
class orm_memoized(object):
"""A read-only @property that is only evaluated once.
Adds a listener so that values are expired
when the object is part of a flush.
"""
all_memoized = collections.defaultdict(set)
def __init__(self, fget, doc=None):
self.fget = fget
self.__doc__ = doc or fget.__doc__
self.__name__ = fget.__name__
def __get__(self, obj, cls):
if obj is None:
return self
self.all_memoized[cls].add(self.__name__)
obj.__dict__[self.__name__] = result = self.fget(obj)
return result
@event.listens_for(Session, "after_flush")
def expire(session, flush_context):
"""expire all orm_memoized en masse for a given flush"""
for obj in session.identity_map.values():
for name in orm_memoized.all_memoized[obj.__class__]:
obj.__dict__.pop(name, None)
Base= declarative_base()
class MyData(Base):
__tablename__ = "mydata"
id = Column(Integer, primary_key=True)
related = relationship("MyRelated")
@orm_memoized
def related_count(self):
return object_session(self).\
query(func.count(MyRelated.id)).\
with_parent(self).scalar()
class MyRelated(Base):
__tablename__ = 'myrelated'
id = Column(Integer, primary_key=True)
parent_id = Column(Integer, ForeignKey('mydata.id'))
e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)
s = Session(e)
s.add_all([
MyData(related=[
MyRelated(),
MyRelated(),
MyRelated(),
MyRelated(),
MyRelated(),
])
])
s.commit()
md = s.query(MyData).first()
assert md.related_count == 5
assert md.related_count == 5
rel = s.query(MyRelated)[0:3]
for mr in rel:
s.delete(mr)
s.flush()
assert md.related_count == 2