1. Michael Bayer
  2. sqlalchemy

Wiki

Clone wiki

sqlalchemy / UsageRecipes / ExpiryMemoized

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:

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:

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

Updated