observer.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376
  1. """
  2. This module provides a decorator function for observing changes in a given
  3. property. Internally the decorator is implemented using SQLAlchemy event
  4. listeners. Both column properties and relationship properties can be observed.
  5. Property observers can be used for pre-calculating aggregates and automatic
  6. real-time data denormalization.
  7. Simple observers
  8. ----------------
  9. At the heart of the observer extension is the :func:`observes` decorator. You
  10. mark some property path as being observed and the marked method will get
  11. notified when any changes are made to given path.
  12. Consider the following model structure:
  13. ::
  14. class Director(Base):
  15. __tablename__ = 'director'
  16. id = sa.Column(sa.Integer, primary_key=True)
  17. name = sa.Column(sa.String)
  18. date_of_birth = sa.Column(sa.Date)
  19. class Movie(Base):
  20. __tablename__ = 'movie'
  21. id = sa.Column(sa.Integer, primary_key=True)
  22. name = sa.Column(sa.String)
  23. director_id = sa.Column(sa.Integer, sa.ForeignKey(Director.id))
  24. director = sa.orm.relationship(Director, backref='movies')
  25. Now consider we want to show movies in some listing ordered by director id
  26. first and movie id secondly. If we have many movies then using joins and
  27. ordering by Director.name will be very slow. Here is where denormalization
  28. and :func:`observes` comes to rescue the day. Let's add a new column called
  29. director_name to Movie which will get automatically copied from associated
  30. Director.
  31. ::
  32. from sqlalchemy_utils import observes
  33. class Movie(Base):
  34. # same as before..
  35. director_name = sa.Column(sa.String)
  36. @observes('director')
  37. def director_observer(self, director):
  38. self.director_name = director.name
  39. .. note::
  40. This example could be done much more efficiently using a compound foreign
  41. key from director_name, director_id to Director.name, Director.id but for
  42. the sake of simplicity we added this as an example.
  43. Observes vs aggregated
  44. ----------------------
  45. :func:`observes` and :func:`.aggregates.aggregated` can be used for similar
  46. things. However performance wise you should take the following things into
  47. consideration:
  48. * :func:`observes` works always inside transaction and deals with objects. If
  49. the relationship observer is observing has a large number of objects it's
  50. better to use :func:`.aggregates.aggregated`.
  51. * :func:`.aggregates.aggregated` always executes one additional query per
  52. aggregate so in scenarios where the observed relationship has only a handful
  53. of objects it's better to use :func:`observes` instead.
  54. Example 1. Movie with many ratings
  55. Let's say we have a Movie object with potentially thousands of ratings. In this
  56. case we should always use :func:`.aggregates.aggregated` since iterating
  57. through thousands of objects is slow and very memory consuming.
  58. Example 2. Product with denormalized catalog name
  59. Each product belongs to one catalog. Here it is natural to use :func:`observes`
  60. for data denormalization.
  61. Deeply nested observing
  62. -----------------------
  63. Consider the following model structure where Catalog has many Categories and
  64. Category has many Products.
  65. ::
  66. class Catalog(Base):
  67. __tablename__ = 'catalog'
  68. id = sa.Column(sa.Integer, primary_key=True)
  69. product_count = sa.Column(sa.Integer, default=0)
  70. @observes('categories.products')
  71. def product_observer(self, products):
  72. self.product_count = len(products)
  73. categories = sa.orm.relationship('Category', backref='catalog')
  74. class Category(Base):
  75. __tablename__ = 'category'
  76. id = sa.Column(sa.Integer, primary_key=True)
  77. catalog_id = sa.Column(sa.Integer, sa.ForeignKey('catalog.id'))
  78. products = sa.orm.relationship('Product', backref='category')
  79. class Product(Base):
  80. __tablename__ = 'product'
  81. id = sa.Column(sa.Integer, primary_key=True)
  82. price = sa.Column(sa.Numeric)
  83. category_id = sa.Column(sa.Integer, sa.ForeignKey('category.id'))
  84. :func:`observes` is smart enough to:
  85. * Notify catalog objects of any changes in associated Product objects
  86. * Notify catalog objects of any changes in Category objects that affect
  87. products (for example if Category gets deleted, or a new Category is added to
  88. Catalog with any number of Products)
  89. ::
  90. category = Category(
  91. products=[Product(), Product()]
  92. )
  93. category2 = Category(
  94. product=[Product()]
  95. )
  96. catalog = Catalog(
  97. categories=[category, category2]
  98. )
  99. session.add(catalog)
  100. session.commit()
  101. catalog.product_count # 2
  102. session.delete(category)
  103. session.commit()
  104. catalog.product_count # 1
  105. Observing multiple columns
  106. -----------------------
  107. You can also observe multiple columns by specifying all the observable columns
  108. in the decorator.
  109. ::
  110. class Order(Base):
  111. __tablename__ = 'order'
  112. id = sa.Column(sa.Integer, primary_key=True)
  113. unit_price = sa.Column(sa.Integer)
  114. amount = sa.Column(sa.Integer)
  115. total_price = sa.Column(sa.Integer)
  116. @observes('amount', 'unit_price')
  117. def total_price_observer(self, amount, unit_price):
  118. self.total_price = amount * unit_price
  119. """
  120. import itertools
  121. from collections import defaultdict, namedtuple
  122. from collections.abc import Iterable
  123. import sqlalchemy as sa
  124. from .functions import getdotattr, has_changes
  125. from .path import AttrPath
  126. from .utils import is_sequence
  127. Callback = namedtuple('Callback', ['func', 'backref', 'fullpath'])
  128. class PropertyObserver(object):
  129. def __init__(self):
  130. self.listener_args = [
  131. (
  132. sa.orm.mapper,
  133. 'mapper_configured',
  134. self.update_generator_registry
  135. ),
  136. (
  137. sa.orm.mapper,
  138. 'after_configured',
  139. self.gather_paths
  140. ),
  141. (
  142. sa.orm.session.Session,
  143. 'before_flush',
  144. self.invoke_callbacks
  145. )
  146. ]
  147. self.callback_map = defaultdict(list)
  148. # TODO: make the registry a WeakKey dict
  149. self.generator_registry = defaultdict(list)
  150. def remove_listeners(self):
  151. for args in self.listener_args:
  152. sa.event.remove(*args)
  153. def register_listeners(self):
  154. for args in self.listener_args:
  155. if not sa.event.contains(*args):
  156. sa.event.listen(*args)
  157. def __repr__(self):
  158. return '<PropertyObserver>'
  159. def update_generator_registry(self, mapper, class_):
  160. """
  161. Adds generator functions to generator_registry.
  162. """
  163. for generator in class_.__dict__.values():
  164. if hasattr(generator, '__observes__'):
  165. self.generator_registry[class_].append(
  166. generator
  167. )
  168. def gather_paths(self):
  169. for class_, generators in self.generator_registry.items():
  170. for callback in generators:
  171. full_paths = []
  172. for call_path in callback.__observes__:
  173. full_paths.append(AttrPath(class_, call_path))
  174. for path in full_paths:
  175. self.callback_map[class_].append(
  176. Callback(
  177. func=callback,
  178. backref=None,
  179. fullpath=full_paths
  180. )
  181. )
  182. for index in range(len(path)):
  183. i = index + 1
  184. prop = path[index].property
  185. if isinstance(prop, sa.orm.RelationshipProperty):
  186. prop_class = path[index].property.mapper.class_
  187. self.callback_map[prop_class].append(
  188. Callback(
  189. func=callback,
  190. backref=~ (path[:i]),
  191. fullpath=full_paths
  192. )
  193. )
  194. def gather_callback_args(self, obj, callbacks):
  195. session = sa.orm.object_session(obj)
  196. for callback in callbacks:
  197. backref = callback.backref
  198. root_objs = getdotattr(obj, backref) if backref else obj
  199. if root_objs:
  200. if not isinstance(root_objs, Iterable):
  201. root_objs = [root_objs]
  202. with session.no_autoflush:
  203. for root_obj in root_objs:
  204. if root_obj:
  205. args = self.get_callback_args(root_obj, callback)
  206. if args:
  207. yield args
  208. def get_callback_args(self, root_obj, callback):
  209. session = sa.orm.object_session(root_obj)
  210. objects = [getdotattr(
  211. root_obj,
  212. path,
  213. lambda obj: obj not in session.deleted
  214. ) for path in callback.fullpath]
  215. paths = [str(path) for path in callback.fullpath]
  216. for path in paths:
  217. if '.' in path or has_changes(root_obj, path):
  218. return (
  219. root_obj,
  220. callback.func,
  221. objects
  222. )
  223. def iterate_objects_and_callbacks(self, session):
  224. objs = itertools.chain(session.new, session.dirty, session.deleted)
  225. for obj in objs:
  226. for class_, callbacks in self.callback_map.items():
  227. if isinstance(obj, class_):
  228. yield obj, callbacks
  229. def invoke_callbacks(self, session, ctx, instances):
  230. callback_args = defaultdict(lambda: defaultdict(set))
  231. for obj, callbacks in self.iterate_objects_and_callbacks(session):
  232. args = self.gather_callback_args(obj, callbacks)
  233. for (root_obj, func, objects) in args:
  234. if not callback_args[root_obj][func]:
  235. callback_args[root_obj][func] = {}
  236. for i, object_ in enumerate(objects):
  237. if is_sequence(object_):
  238. callback_args[root_obj][func][i] = (
  239. callback_args[root_obj][func].get(i, set()) |
  240. set(object_)
  241. )
  242. else:
  243. callback_args[root_obj][func][i] = object_
  244. for root_obj, callback_objs in callback_args.items():
  245. for callback, objs in callback_objs.items():
  246. callback(root_obj, *[objs[i] for i in range(len(objs))])
  247. observer = PropertyObserver()
  248. def observes(*paths, **observer_kw):
  249. """
  250. Mark method as property observer for the given property path. Inside
  251. transaction observer gathers all changes made in given property path and
  252. feeds the changed objects to observer-marked method at the before flush
  253. phase.
  254. ::
  255. from sqlalchemy_utils import observes
  256. class Catalog(Base):
  257. __tablename__ = 'catalog'
  258. id = sa.Column(sa.Integer, primary_key=True)
  259. category_count = sa.Column(sa.Integer, default=0)
  260. @observes('categories')
  261. def category_observer(self, categories):
  262. self.category_count = len(categories)
  263. class Category(Base):
  264. __tablename__ = 'category'
  265. id = sa.Column(sa.Integer, primary_key=True)
  266. catalog_id = sa.Column(sa.Integer, sa.ForeignKey('catalog.id'))
  267. catalog = Catalog(categories=[Category(), Category()])
  268. session.add(catalog)
  269. session.commit()
  270. catalog.category_count # 2
  271. .. versionadded: 0.28.0
  272. :param *paths: One or more dot-notated property paths, eg.
  273. 'categories.products.price'
  274. :param **observer: A dictionary where value for key 'observer' contains
  275. :meth:`PropertyObserver` object
  276. """
  277. observer_ = observer_kw.pop('observer', observer)
  278. observer_.register_listeners()
  279. def wraps(func):
  280. def wrapper(self, *args, **kwargs):
  281. return func(self, *args, **kwargs)
  282. wrapper.__observes__ = paths
  283. return wrapper
  284. return wraps