[3] | 1 | # orm/unitofwork.py |
---|
| 2 | # Copyright (C) 2005, 2006, 2007, 2008, 2009 Michael Bayer mike_mp@zzzcomputing.com |
---|
| 3 | # |
---|
| 4 | # This module is part of SQLAlchemy and is released under |
---|
| 5 | # the MIT License: http://www.opensource.org/licenses/mit-license.php |
---|
| 6 | |
---|
| 7 | """The internals for the Unit Of Work system. |
---|
| 8 | |
---|
| 9 | Includes hooks into the attributes package enabling the routing of |
---|
| 10 | change events to Unit Of Work objects, as well as the flush() |
---|
| 11 | mechanism which creates a dependency structure that executes change |
---|
| 12 | operations. |
---|
| 13 | |
---|
| 14 | A Unit of Work is essentially a system of maintaining a graph of |
---|
| 15 | in-memory objects and their modified state. Objects are maintained as |
---|
| 16 | unique against their primary key identity using an *identity map* |
---|
| 17 | pattern. The Unit of Work then maintains lists of objects that are |
---|
| 18 | new, dirty, or deleted and provides the capability to flush all those |
---|
| 19 | changes at once. |
---|
| 20 | |
---|
| 21 | """ |
---|
| 22 | |
---|
| 23 | from sqlalchemy import util, log, topological |
---|
| 24 | from sqlalchemy.orm import attributes, interfaces |
---|
| 25 | from sqlalchemy.orm import util as mapperutil |
---|
| 26 | from sqlalchemy.orm.mapper import _state_mapper |
---|
| 27 | |
---|
| 28 | # Load lazily |
---|
| 29 | object_session = None |
---|
| 30 | _state_session = None |
---|
| 31 | |
---|
| 32 | class UOWEventHandler(interfaces.AttributeExtension): |
---|
| 33 | """An event handler added to all relation attributes which handles |
---|
| 34 | session cascade operations. |
---|
| 35 | """ |
---|
| 36 | |
---|
| 37 | active_history = False |
---|
| 38 | |
---|
| 39 | def __init__(self, key): |
---|
| 40 | self.key = key |
---|
| 41 | |
---|
| 42 | def append(self, state, item, initiator): |
---|
| 43 | # process "save_update" cascade rules for when an instance is appended to the list of another instance |
---|
| 44 | sess = _state_session(state) |
---|
| 45 | if sess: |
---|
| 46 | prop = _state_mapper(state).get_property(self.key) |
---|
| 47 | if prop.cascade.save_update and item not in sess: |
---|
| 48 | sess.add(item) |
---|
| 49 | return item |
---|
| 50 | |
---|
| 51 | def remove(self, state, item, initiator): |
---|
| 52 | sess = _state_session(state) |
---|
| 53 | if sess: |
---|
| 54 | prop = _state_mapper(state).get_property(self.key) |
---|
| 55 | # expunge pending orphans |
---|
| 56 | if prop.cascade.delete_orphan and \ |
---|
| 57 | item in sess.new and \ |
---|
| 58 | prop.mapper._is_orphan(attributes.instance_state(item)): |
---|
| 59 | sess.expunge(item) |
---|
| 60 | |
---|
| 61 | def set(self, state, newvalue, oldvalue, initiator): |
---|
| 62 | # process "save_update" cascade rules for when an instance is attached to another instance |
---|
| 63 | if oldvalue is newvalue: |
---|
| 64 | return newvalue |
---|
| 65 | sess = _state_session(state) |
---|
| 66 | if sess: |
---|
| 67 | prop = _state_mapper(state).get_property(self.key) |
---|
| 68 | if newvalue is not None and prop.cascade.save_update and newvalue not in sess: |
---|
| 69 | sess.add(newvalue) |
---|
| 70 | if prop.cascade.delete_orphan and oldvalue in sess.new and \ |
---|
| 71 | prop.mapper._is_orphan(attributes.instance_state(oldvalue)): |
---|
| 72 | sess.expunge(oldvalue) |
---|
| 73 | return newvalue |
---|
| 74 | |
---|
| 75 | |
---|
| 76 | class UOWTransaction(object): |
---|
| 77 | """Handles the details of organizing and executing transaction |
---|
| 78 | tasks during a UnitOfWork object's flush() operation. |
---|
| 79 | |
---|
| 80 | The central operation is to form a graph of nodes represented by the |
---|
| 81 | ``UOWTask`` class, which is then traversed by a ``UOWExecutor`` object |
---|
| 82 | that issues SQL and instance-synchronizing operations via the related |
---|
| 83 | packages. |
---|
| 84 | """ |
---|
| 85 | |
---|
| 86 | def __init__(self, session): |
---|
| 87 | self.session = session |
---|
| 88 | self.mapper_flush_opts = session._mapper_flush_opts |
---|
| 89 | |
---|
| 90 | # stores tuples of mapper/dependent mapper pairs, |
---|
| 91 | # representing a partial ordering fed into topological sort |
---|
| 92 | self.dependencies = set() |
---|
| 93 | |
---|
| 94 | # dictionary of mappers to UOWTasks |
---|
| 95 | self.tasks = {} |
---|
| 96 | |
---|
| 97 | # dictionary used by external actors to store arbitrary state |
---|
| 98 | # information. |
---|
| 99 | self.attributes = {} |
---|
| 100 | |
---|
| 101 | self.processors = set() |
---|
| 102 | |
---|
| 103 | def get_attribute_history(self, state, key, passive=True): |
---|
| 104 | hashkey = ("history", state, key) |
---|
| 105 | |
---|
| 106 | # cache the objects, not the states; the strong reference here |
---|
| 107 | # prevents newly loaded objects from being dereferenced during the |
---|
| 108 | # flush process |
---|
| 109 | if hashkey in self.attributes: |
---|
| 110 | (history, cached_passive) = self.attributes[hashkey] |
---|
| 111 | # if the cached lookup was "passive" and now we want non-passive, do a non-passive |
---|
| 112 | # lookup and re-cache |
---|
| 113 | if cached_passive and not passive: |
---|
| 114 | history = attributes.get_state_history(state, key, passive=False) |
---|
| 115 | self.attributes[hashkey] = (history, passive) |
---|
| 116 | else: |
---|
| 117 | history = attributes.get_state_history(state, key, passive=passive) |
---|
| 118 | self.attributes[hashkey] = (history, passive) |
---|
| 119 | |
---|
| 120 | if not history or not state.get_impl(key).uses_objects: |
---|
| 121 | return history |
---|
| 122 | else: |
---|
| 123 | return history.as_state() |
---|
| 124 | |
---|
| 125 | def register_object(self, state, isdelete=False, listonly=False, postupdate=False, post_update_cols=None): |
---|
| 126 | |
---|
| 127 | # if object is not in the overall session, do nothing |
---|
| 128 | if not self.session._contains_state(state): |
---|
| 129 | if self._should_log_debug: |
---|
| 130 | self.logger.debug("object %s not part of session, not registering for flush" % |
---|
| 131 | (mapperutil.state_str(state))) |
---|
| 132 | return |
---|
| 133 | |
---|
| 134 | if self._should_log_debug: |
---|
| 135 | self.logger.debug("register object for flush: %s isdelete=%s listonly=%s postupdate=%s" |
---|
| 136 | % (mapperutil.state_str(state), isdelete, listonly, postupdate)) |
---|
| 137 | |
---|
| 138 | mapper = _state_mapper(state) |
---|
| 139 | |
---|
| 140 | task = self.get_task_by_mapper(mapper) |
---|
| 141 | if postupdate: |
---|
| 142 | task.append_postupdate(state, post_update_cols) |
---|
| 143 | else: |
---|
| 144 | task.append(state, listonly=listonly, isdelete=isdelete) |
---|
| 145 | |
---|
| 146 | # ensure the mapper for this object has had its |
---|
| 147 | # DependencyProcessors added. |
---|
| 148 | if mapper not in self.processors: |
---|
| 149 | mapper._register_processors(self) |
---|
| 150 | self.processors.add(mapper) |
---|
| 151 | |
---|
| 152 | if mapper.base_mapper not in self.processors: |
---|
| 153 | mapper.base_mapper._register_processors(self) |
---|
| 154 | self.processors.add(mapper.base_mapper) |
---|
| 155 | |
---|
| 156 | def set_row_switch(self, state): |
---|
| 157 | """mark a deleted object as a 'row switch'. |
---|
| 158 | |
---|
| 159 | this indicates that an INSERT statement elsewhere corresponds to this DELETE; |
---|
| 160 | the INSERT is converted to an UPDATE and the DELETE does not occur. |
---|
| 161 | |
---|
| 162 | """ |
---|
| 163 | mapper = _state_mapper(state) |
---|
| 164 | task = self.get_task_by_mapper(mapper) |
---|
| 165 | taskelement = task._objects[state] |
---|
| 166 | taskelement.isdelete = "rowswitch" |
---|
| 167 | |
---|
| 168 | def is_deleted(self, state): |
---|
| 169 | """return true if the given state is marked as deleted within this UOWTransaction.""" |
---|
| 170 | |
---|
| 171 | mapper = _state_mapper(state) |
---|
| 172 | task = self.get_task_by_mapper(mapper) |
---|
| 173 | return task.is_deleted(state) |
---|
| 174 | |
---|
| 175 | def get_task_by_mapper(self, mapper, dontcreate=False): |
---|
| 176 | """return UOWTask element corresponding to the given mapper. |
---|
| 177 | |
---|
| 178 | Will create a new UOWTask, including a UOWTask corresponding to the |
---|
| 179 | "base" inherited mapper, if needed, unless the dontcreate flag is True. |
---|
| 180 | |
---|
| 181 | """ |
---|
| 182 | try: |
---|
| 183 | return self.tasks[mapper] |
---|
| 184 | except KeyError: |
---|
| 185 | if dontcreate: |
---|
| 186 | return None |
---|
| 187 | |
---|
| 188 | base_mapper = mapper.base_mapper |
---|
| 189 | if base_mapper in self.tasks: |
---|
| 190 | base_task = self.tasks[base_mapper] |
---|
| 191 | else: |
---|
| 192 | self.tasks[base_mapper] = base_task = UOWTask(self, base_mapper) |
---|
| 193 | base_mapper._register_dependencies(self) |
---|
| 194 | |
---|
| 195 | if mapper not in self.tasks: |
---|
| 196 | self.tasks[mapper] = task = UOWTask(self, mapper, base_task=base_task) |
---|
| 197 | mapper._register_dependencies(self) |
---|
| 198 | else: |
---|
| 199 | task = self.tasks[mapper] |
---|
| 200 | |
---|
| 201 | return task |
---|
| 202 | |
---|
| 203 | def register_dependency(self, mapper, dependency): |
---|
| 204 | """register a dependency between two mappers. |
---|
| 205 | |
---|
| 206 | Called by ``mapper.PropertyLoader`` to register the objects |
---|
| 207 | handled by one mapper being dependent on the objects handled |
---|
| 208 | by another. |
---|
| 209 | |
---|
| 210 | """ |
---|
| 211 | # correct for primary mapper |
---|
| 212 | # also convert to the "base mapper", the parentmost task at the top of an inheritance chain |
---|
| 213 | # dependency sorting is done via non-inheriting mappers only, dependencies between mappers |
---|
| 214 | # in the same inheritance chain is done at the per-object level |
---|
| 215 | mapper = mapper.primary_mapper().base_mapper |
---|
| 216 | dependency = dependency.primary_mapper().base_mapper |
---|
| 217 | |
---|
| 218 | self.dependencies.add((mapper, dependency)) |
---|
| 219 | |
---|
| 220 | def register_processor(self, mapper, processor, mapperfrom): |
---|
| 221 | """register a dependency processor, corresponding to |
---|
| 222 | operations which occur between two mappers. |
---|
| 223 | |
---|
| 224 | """ |
---|
| 225 | # correct for primary mapper |
---|
| 226 | mapper = mapper.primary_mapper() |
---|
| 227 | mapperfrom = mapperfrom.primary_mapper() |
---|
| 228 | |
---|
| 229 | task = self.get_task_by_mapper(mapper) |
---|
| 230 | targettask = self.get_task_by_mapper(mapperfrom) |
---|
| 231 | up = UOWDependencyProcessor(processor, targettask) |
---|
| 232 | task.dependencies.add(up) |
---|
| 233 | |
---|
| 234 | def execute(self): |
---|
| 235 | """Execute this UOWTransaction. |
---|
| 236 | |
---|
| 237 | This will organize all collected UOWTasks into a dependency-sorted |
---|
| 238 | list which is then traversed using the traversal scheme |
---|
| 239 | encoded in the UOWExecutor class. Operations to mappers and dependency |
---|
| 240 | processors are fired off in order to issue SQL to the database and |
---|
| 241 | synchronize instance attributes with database values and related |
---|
| 242 | foreign key values.""" |
---|
| 243 | |
---|
| 244 | # pre-execute dependency processors. this process may |
---|
| 245 | # result in new tasks, objects and/or dependency processors being added, |
---|
| 246 | # particularly with 'delete-orphan' cascade rules. |
---|
| 247 | # keep running through the full list of tasks until all |
---|
| 248 | # objects have been processed. |
---|
| 249 | while True: |
---|
| 250 | ret = False |
---|
| 251 | for task in self.tasks.values(): |
---|
| 252 | for up in list(task.dependencies): |
---|
| 253 | if up.preexecute(self): |
---|
| 254 | ret = True |
---|
| 255 | if not ret: |
---|
| 256 | break |
---|
| 257 | |
---|
| 258 | tasks = self._sort_dependencies() |
---|
| 259 | if self._should_log_info: |
---|
| 260 | self.logger.info("Task dump:\n" + self._dump(tasks)) |
---|
| 261 | UOWExecutor().execute(self, tasks) |
---|
| 262 | if self._should_log_info: |
---|
| 263 | self.logger.info("Execute Complete") |
---|
| 264 | |
---|
| 265 | def _dump(self, tasks): |
---|
| 266 | from uowdumper import UOWDumper |
---|
| 267 | return UOWDumper.dump(tasks) |
---|
| 268 | |
---|
| 269 | @property |
---|
| 270 | def elements(self): |
---|
| 271 | """Iterate UOWTaskElements.""" |
---|
| 272 | |
---|
| 273 | for task in self.tasks.itervalues(): |
---|
| 274 | for elem in task.elements: |
---|
| 275 | yield elem |
---|
| 276 | |
---|
| 277 | def finalize_flush_changes(self): |
---|
| 278 | """mark processed objects as clean / deleted after a successful flush(). |
---|
| 279 | |
---|
| 280 | this method is called within the flush() method after the |
---|
| 281 | execute() method has succeeded and the transaction has been committed. |
---|
| 282 | """ |
---|
| 283 | |
---|
| 284 | for elem in self.elements: |
---|
| 285 | if elem.isdelete: |
---|
| 286 | self.session._remove_newly_deleted(elem.state) |
---|
| 287 | elif not elem.listonly: |
---|
| 288 | self.session._register_newly_persistent(elem.state) |
---|
| 289 | |
---|
| 290 | def _sort_dependencies(self): |
---|
| 291 | nodes = topological.sort_with_cycles(self.dependencies, |
---|
| 292 | [t.mapper for t in self.tasks.itervalues() if t.base_task is t] |
---|
| 293 | ) |
---|
| 294 | |
---|
| 295 | ret = [] |
---|
| 296 | for item, cycles in nodes: |
---|
| 297 | task = self.get_task_by_mapper(item) |
---|
| 298 | if cycles: |
---|
| 299 | for t in task._sort_circular_dependencies(self, [self.get_task_by_mapper(i) for i in cycles]): |
---|
| 300 | ret.append(t) |
---|
| 301 | else: |
---|
| 302 | ret.append(task) |
---|
| 303 | |
---|
| 304 | if self._should_log_debug: |
---|
| 305 | self.logger.debug("Dependent tuples:\n" + "\n".join( |
---|
| 306 | "(%s->%s)" % (d[0].class_.__name__, d[1].class_.__name__) |
---|
| 307 | for d in self.dependencies)) |
---|
| 308 | self.logger.debug("Dependency sort:\n"+ str(ret)) |
---|
| 309 | return ret |
---|
| 310 | |
---|
| 311 | log.class_logger(UOWTransaction) |
---|
| 312 | |
---|
| 313 | class UOWTask(object): |
---|
| 314 | """A collection of mapped states corresponding to a particular mapper.""" |
---|
| 315 | |
---|
| 316 | def __init__(self, uowtransaction, mapper, base_task=None): |
---|
| 317 | self.uowtransaction = uowtransaction |
---|
| 318 | |
---|
| 319 | # base_task is the UOWTask which represents the "base mapper" |
---|
| 320 | # in our mapper's inheritance chain. if the mapper does not |
---|
| 321 | # inherit from any other mapper, the base_task is self. |
---|
| 322 | # the _inheriting_tasks dictionary is a dictionary present only |
---|
| 323 | # on the "base_task"-holding UOWTask, which maps all mappers within |
---|
| 324 | # an inheritance hierarchy to their corresponding UOWTask instances. |
---|
| 325 | if base_task is None: |
---|
| 326 | self.base_task = self |
---|
| 327 | self._inheriting_tasks = {mapper:self} |
---|
| 328 | else: |
---|
| 329 | self.base_task = base_task |
---|
| 330 | base_task._inheriting_tasks[mapper] = self |
---|
| 331 | |
---|
| 332 | # the Mapper which this UOWTask corresponds to |
---|
| 333 | self.mapper = mapper |
---|
| 334 | |
---|
| 335 | # mapping of InstanceState -> UOWTaskElement |
---|
| 336 | self._objects = {} |
---|
| 337 | |
---|
| 338 | self.dependent_tasks = [] |
---|
| 339 | self.dependencies = set() |
---|
| 340 | self.cyclical_dependencies = set() |
---|
| 341 | |
---|
| 342 | @util.memoized_property |
---|
| 343 | def inheriting_mappers(self): |
---|
| 344 | return list(self.mapper.polymorphic_iterator()) |
---|
| 345 | |
---|
| 346 | @property |
---|
| 347 | def polymorphic_tasks(self): |
---|
| 348 | """Return an iterator of UOWTask objects corresponding to the |
---|
| 349 | inheritance sequence of this UOWTask's mapper. |
---|
| 350 | |
---|
| 351 | e.g. if mapper B and mapper C inherit from mapper A, and |
---|
| 352 | mapper D inherits from B: |
---|
| 353 | |
---|
| 354 | mapperA -> mapperB -> mapperD |
---|
| 355 | -> mapperC |
---|
| 356 | |
---|
| 357 | the inheritance sequence starting at mapper A is a depth-first |
---|
| 358 | traversal: |
---|
| 359 | |
---|
| 360 | [mapperA, mapperB, mapperD, mapperC] |
---|
| 361 | |
---|
| 362 | this method will therefore return |
---|
| 363 | |
---|
| 364 | [UOWTask(mapperA), UOWTask(mapperB), UOWTask(mapperD), |
---|
| 365 | UOWTask(mapperC)] |
---|
| 366 | |
---|
| 367 | The concept of "polymporphic iteration" is adapted into |
---|
| 368 | several property-based iterators which return object |
---|
| 369 | instances, UOWTaskElements and UOWDependencyProcessors in an |
---|
| 370 | order corresponding to this sequence of parent UOWTasks. This |
---|
| 371 | is used to issue operations related to inheritance-chains of |
---|
| 372 | mappers in the proper order based on dependencies between |
---|
| 373 | those mappers. |
---|
| 374 | |
---|
| 375 | """ |
---|
| 376 | for mapper in self.inheriting_mappers: |
---|
| 377 | t = self.base_task._inheriting_tasks.get(mapper, None) |
---|
| 378 | if t is not None: |
---|
| 379 | yield t |
---|
| 380 | |
---|
| 381 | def is_empty(self): |
---|
| 382 | """return True if this UOWTask is 'empty', meaning it has no child items. |
---|
| 383 | |
---|
| 384 | used only for debugging output. |
---|
| 385 | """ |
---|
| 386 | |
---|
| 387 | return not self._objects and not self.dependencies |
---|
| 388 | |
---|
| 389 | def append(self, state, listonly=False, isdelete=False): |
---|
| 390 | if state not in self._objects: |
---|
| 391 | self._objects[state] = rec = UOWTaskElement(state) |
---|
| 392 | else: |
---|
| 393 | rec = self._objects[state] |
---|
| 394 | |
---|
| 395 | rec.update(listonly, isdelete) |
---|
| 396 | |
---|
| 397 | def append_postupdate(self, state, post_update_cols): |
---|
| 398 | """issue a 'post update' UPDATE statement via this object's mapper immediately. |
---|
| 399 | |
---|
| 400 | this operation is used only with relations that specify the `post_update=True` |
---|
| 401 | flag. |
---|
| 402 | """ |
---|
| 403 | |
---|
| 404 | # postupdates are UPDATED immeditely (for now) |
---|
| 405 | # convert post_update_cols list to a Set so that __hash__() is used to compare columns |
---|
| 406 | # instead of __eq__() |
---|
| 407 | self.mapper._save_obj([state], self.uowtransaction, postupdate=True, post_update_cols=set(post_update_cols)) |
---|
| 408 | |
---|
| 409 | def __contains__(self, state): |
---|
| 410 | """return True if the given object is contained within this UOWTask or inheriting tasks.""" |
---|
| 411 | |
---|
| 412 | for task in self.polymorphic_tasks: |
---|
| 413 | if state in task._objects: |
---|
| 414 | return True |
---|
| 415 | else: |
---|
| 416 | return False |
---|
| 417 | |
---|
| 418 | def is_deleted(self, state): |
---|
| 419 | """return True if the given object is marked as to be deleted within this UOWTask.""" |
---|
| 420 | |
---|
| 421 | try: |
---|
| 422 | return self._objects[state].isdelete |
---|
| 423 | except KeyError: |
---|
| 424 | return False |
---|
| 425 | |
---|
| 426 | def _polymorphic_collection(fn): |
---|
| 427 | """return a property that will adapt the collection returned by the |
---|
| 428 | given callable into a polymorphic traversal.""" |
---|
| 429 | |
---|
| 430 | @property |
---|
| 431 | def collection(self): |
---|
| 432 | for task in self.polymorphic_tasks: |
---|
| 433 | for rec in fn(task): |
---|
| 434 | yield rec |
---|
| 435 | return collection |
---|
| 436 | |
---|
| 437 | def _polymorphic_collection_filtered(fn): |
---|
| 438 | |
---|
| 439 | def collection(self, mappers): |
---|
| 440 | for task in self.polymorphic_tasks: |
---|
| 441 | if task.mapper in mappers: |
---|
| 442 | for rec in fn(task): |
---|
| 443 | yield rec |
---|
| 444 | return collection |
---|
| 445 | |
---|
| 446 | @property |
---|
| 447 | def elements(self): |
---|
| 448 | return self._objects.values() |
---|
| 449 | |
---|
| 450 | @_polymorphic_collection |
---|
| 451 | def polymorphic_elements(self): |
---|
| 452 | return self.elements |
---|
| 453 | |
---|
| 454 | @_polymorphic_collection_filtered |
---|
| 455 | def filter_polymorphic_elements(self): |
---|
| 456 | return self.elements |
---|
| 457 | |
---|
| 458 | @property |
---|
| 459 | def polymorphic_tosave_elements(self): |
---|
| 460 | return [rec for rec in self.polymorphic_elements if not rec.isdelete] |
---|
| 461 | |
---|
| 462 | @property |
---|
| 463 | def polymorphic_todelete_elements(self): |
---|
| 464 | return [rec for rec in self.polymorphic_elements if rec.isdelete] |
---|
| 465 | |
---|
| 466 | @property |
---|
| 467 | def polymorphic_tosave_objects(self): |
---|
| 468 | return [ |
---|
| 469 | rec.state for rec in self.polymorphic_elements |
---|
| 470 | if rec.state is not None and not rec.listonly and rec.isdelete is False |
---|
| 471 | ] |
---|
| 472 | |
---|
| 473 | @property |
---|
| 474 | def polymorphic_todelete_objects(self): |
---|
| 475 | return [ |
---|
| 476 | rec.state for rec in self.polymorphic_elements |
---|
| 477 | if rec.state is not None and not rec.listonly and rec.isdelete is True |
---|
| 478 | ] |
---|
| 479 | |
---|
| 480 | @_polymorphic_collection |
---|
| 481 | def polymorphic_dependencies(self): |
---|
| 482 | return self.dependencies |
---|
| 483 | |
---|
| 484 | @_polymorphic_collection |
---|
| 485 | def polymorphic_cyclical_dependencies(self): |
---|
| 486 | return self.cyclical_dependencies |
---|
| 487 | |
---|
| 488 | def _sort_circular_dependencies(self, trans, cycles): |
---|
| 489 | """Topologically sort individual entities with row-level dependencies. |
---|
| 490 | |
---|
| 491 | Builds a modified UOWTask structure, and is invoked when the |
---|
| 492 | per-mapper topological structure is found to have cycles. |
---|
| 493 | |
---|
| 494 | """ |
---|
| 495 | dependencies = {} |
---|
| 496 | def set_processor_for_state(state, depprocessor, target_state, isdelete): |
---|
| 497 | if state not in dependencies: |
---|
| 498 | dependencies[state] = {} |
---|
| 499 | tasks = dependencies[state] |
---|
| 500 | if depprocessor not in tasks: |
---|
| 501 | tasks[depprocessor] = UOWDependencyProcessor( |
---|
| 502 | depprocessor.processor, |
---|
| 503 | UOWTask(self.uowtransaction, depprocessor.targettask.mapper) |
---|
| 504 | ) |
---|
| 505 | tasks[depprocessor].targettask.append(target_state, isdelete=isdelete) |
---|
| 506 | |
---|
| 507 | cycles = set(cycles) |
---|
| 508 | def dependency_in_cycles(dep): |
---|
| 509 | proctask = trans.get_task_by_mapper(dep.processor.mapper.base_mapper, True) |
---|
| 510 | targettask = trans.get_task_by_mapper(dep.targettask.mapper.base_mapper, True) |
---|
| 511 | return targettask in cycles and (proctask is not None and proctask in cycles) |
---|
| 512 | |
---|
| 513 | deps_by_targettask = {} |
---|
| 514 | extradeplist = [] |
---|
| 515 | for task in cycles: |
---|
| 516 | for dep in task.polymorphic_dependencies: |
---|
| 517 | if not dependency_in_cycles(dep): |
---|
| 518 | extradeplist.append(dep) |
---|
| 519 | for t in dep.targettask.polymorphic_tasks: |
---|
| 520 | l = deps_by_targettask.setdefault(t, []) |
---|
| 521 | l.append(dep) |
---|
| 522 | |
---|
| 523 | object_to_original_task = {} |
---|
| 524 | tuples = [] |
---|
| 525 | |
---|
| 526 | for task in cycles: |
---|
| 527 | for subtask in task.polymorphic_tasks: |
---|
| 528 | for taskelement in subtask.elements: |
---|
| 529 | state = taskelement.state |
---|
| 530 | object_to_original_task[state] = subtask |
---|
| 531 | if subtask not in deps_by_targettask: |
---|
| 532 | continue |
---|
| 533 | for dep in deps_by_targettask[subtask]: |
---|
| 534 | if dep.processor.no_dependencies or not dependency_in_cycles(dep): |
---|
| 535 | continue |
---|
| 536 | (processor, targettask) = (dep.processor, dep.targettask) |
---|
| 537 | isdelete = taskelement.isdelete |
---|
| 538 | |
---|
| 539 | # list of dependent objects from this object |
---|
| 540 | (added, unchanged, deleted) = dep.get_object_dependencies(state, trans, passive=True) |
---|
| 541 | if not added and not unchanged and not deleted: |
---|
| 542 | continue |
---|
| 543 | |
---|
| 544 | # the task corresponding to saving/deleting of those dependent objects |
---|
| 545 | childtask = trans.get_task_by_mapper(processor.mapper) |
---|
| 546 | |
---|
| 547 | childlist = added + unchanged + deleted |
---|
| 548 | |
---|
| 549 | for o in childlist: |
---|
| 550 | if o is None: |
---|
| 551 | continue |
---|
| 552 | |
---|
| 553 | if o not in childtask: |
---|
| 554 | childtask.append(o, listonly=True) |
---|
| 555 | object_to_original_task[o] = childtask |
---|
| 556 | |
---|
| 557 | whosdep = dep.whose_dependent_on_who(state, o) |
---|
| 558 | if whosdep is not None: |
---|
| 559 | tuples.append(whosdep) |
---|
| 560 | |
---|
| 561 | if whosdep[0] is state: |
---|
| 562 | set_processor_for_state(whosdep[0], dep, whosdep[0], isdelete=isdelete) |
---|
| 563 | else: |
---|
| 564 | set_processor_for_state(whosdep[0], dep, whosdep[1], isdelete=isdelete) |
---|
| 565 | else: |
---|
| 566 | # TODO: no test coverage here |
---|
| 567 | set_processor_for_state(state, dep, state, isdelete=isdelete) |
---|
| 568 | |
---|
| 569 | t = UOWTask(self.uowtransaction, self.mapper) |
---|
| 570 | t.dependencies.update(extradeplist) |
---|
| 571 | |
---|
| 572 | used_tasks = set() |
---|
| 573 | |
---|
| 574 | # rationale for "tree" sort as opposed to a straight |
---|
| 575 | # dependency - keep non-dependent objects |
---|
| 576 | # grouped together, so that insert ordering as determined |
---|
| 577 | # by session.add() is maintained. |
---|
| 578 | # An alternative might be to represent the "insert order" |
---|
| 579 | # as part of the topological sort itself, which would |
---|
| 580 | # eliminate the need for this step (but may make the original |
---|
| 581 | # topological sort more expensive) |
---|
| 582 | head = topological.sort_as_tree(tuples, object_to_original_task.iterkeys()) |
---|
| 583 | if head is not None: |
---|
| 584 | original_to_tasks = {} |
---|
| 585 | stack = [(head, t)] |
---|
| 586 | while stack: |
---|
| 587 | ((state, cycles, children), parenttask) = stack.pop() |
---|
| 588 | |
---|
| 589 | originating_task = object_to_original_task[state] |
---|
| 590 | used_tasks.add(originating_task) |
---|
| 591 | |
---|
| 592 | if (parenttask, originating_task) not in original_to_tasks: |
---|
| 593 | task = UOWTask(self.uowtransaction, originating_task.mapper) |
---|
| 594 | original_to_tasks[(parenttask, originating_task)] = task |
---|
| 595 | parenttask.dependent_tasks.append(task) |
---|
| 596 | else: |
---|
| 597 | task = original_to_tasks[(parenttask, originating_task)] |
---|
| 598 | |
---|
| 599 | task.append(state, originating_task._objects[state].listonly, isdelete=originating_task._objects[state].isdelete) |
---|
| 600 | |
---|
| 601 | if state in dependencies: |
---|
| 602 | task.cyclical_dependencies.update(dependencies[state].itervalues()) |
---|
| 603 | |
---|
| 604 | stack += [(n, task) for n in children] |
---|
| 605 | |
---|
| 606 | ret = [t] |
---|
| 607 | |
---|
| 608 | # add tasks that were in the cycle, but didnt get assembled |
---|
| 609 | # into the cyclical tree, to the start of the list |
---|
| 610 | for t2 in cycles: |
---|
| 611 | if t2 not in used_tasks and t2 is not self: |
---|
| 612 | localtask = UOWTask(self.uowtransaction, t2.mapper) |
---|
| 613 | for state in t2.elements: |
---|
| 614 | localtask.append(state, t2.listonly, isdelete=t2._objects[state].isdelete) |
---|
| 615 | for dep in t2.dependencies: |
---|
| 616 | localtask.dependencies.add(dep) |
---|
| 617 | ret.insert(0, localtask) |
---|
| 618 | |
---|
| 619 | return ret |
---|
| 620 | |
---|
| 621 | def __repr__(self): |
---|
| 622 | return ("UOWTask(%s) Mapper: '%r'" % (hex(id(self)), self.mapper)) |
---|
| 623 | |
---|
| 624 | class UOWTaskElement(object): |
---|
| 625 | """Corresponds to a single InstanceState to be saved, deleted, |
---|
| 626 | or otherwise marked as having dependencies. A collection of |
---|
| 627 | UOWTaskElements are held by a UOWTask. |
---|
| 628 | |
---|
| 629 | """ |
---|
| 630 | def __init__(self, state): |
---|
| 631 | self.state = state |
---|
| 632 | self.listonly = True |
---|
| 633 | self.isdelete = False |
---|
| 634 | self.preprocessed = set() |
---|
| 635 | |
---|
| 636 | def update(self, listonly, isdelete): |
---|
| 637 | if not listonly and self.listonly: |
---|
| 638 | self.listonly = False |
---|
| 639 | self.preprocessed.clear() |
---|
| 640 | if isdelete and not self.isdelete: |
---|
| 641 | self.isdelete = True |
---|
| 642 | self.preprocessed.clear() |
---|
| 643 | |
---|
| 644 | def __repr__(self): |
---|
| 645 | return "UOWTaskElement/%d: %s/%d %s" % ( |
---|
| 646 | id(self), |
---|
| 647 | self.state.class_.__name__, |
---|
| 648 | id(self.state.obj()), |
---|
| 649 | (self.listonly and 'listonly' or (self.isdelete and 'delete' or 'save')) |
---|
| 650 | ) |
---|
| 651 | |
---|
| 652 | class UOWDependencyProcessor(object): |
---|
| 653 | """In between the saving and deleting of objects, process |
---|
| 654 | dependent data, such as filling in a foreign key on a child item |
---|
| 655 | from a new primary key, or deleting association rows before a |
---|
| 656 | delete. This object acts as a proxy to a DependencyProcessor. |
---|
| 657 | |
---|
| 658 | """ |
---|
| 659 | def __init__(self, processor, targettask): |
---|
| 660 | self.processor = processor |
---|
| 661 | self.targettask = targettask |
---|
| 662 | prop = processor.prop |
---|
| 663 | |
---|
| 664 | # define a set of mappers which |
---|
| 665 | # will filter the lists of entities |
---|
| 666 | # this UOWDP processes. this allows |
---|
| 667 | # MapperProperties to be overridden |
---|
| 668 | # at least for concrete mappers. |
---|
| 669 | self._mappers = set([ |
---|
| 670 | m |
---|
| 671 | for m in self.processor.parent.polymorphic_iterator() |
---|
| 672 | if m._props[prop.key] is prop |
---|
| 673 | ]).union(self.processor.mapper.polymorphic_iterator()) |
---|
| 674 | |
---|
| 675 | def __repr__(self): |
---|
| 676 | return "UOWDependencyProcessor(%s, %s)" % (str(self.processor), str(self.targettask)) |
---|
| 677 | |
---|
| 678 | def __eq__(self, other): |
---|
| 679 | return other.processor is self.processor and other.targettask is self.targettask |
---|
| 680 | |
---|
| 681 | def __hash__(self): |
---|
| 682 | return hash((self.processor, self.targettask)) |
---|
| 683 | |
---|
| 684 | def preexecute(self, trans): |
---|
| 685 | """preprocess all objects contained within this ``UOWDependencyProcessor``s target task. |
---|
| 686 | |
---|
| 687 | This may locate additional objects which should be part of the |
---|
| 688 | transaction, such as those affected deletes, orphans to be |
---|
| 689 | deleted, etc. |
---|
| 690 | |
---|
| 691 | Once an object is preprocessed, its ``UOWTaskElement`` is marked as processed. If subsequent |
---|
| 692 | changes occur to the ``UOWTaskElement``, its processed flag is reset, and will require processing |
---|
| 693 | again. |
---|
| 694 | |
---|
| 695 | Return True if any objects were preprocessed, or False if no |
---|
| 696 | objects were preprocessed. If True is returned, the parent ``UOWTransaction`` will |
---|
| 697 | ultimately call ``preexecute()`` again on all processors until no new objects are processed. |
---|
| 698 | """ |
---|
| 699 | |
---|
| 700 | def getobj(elem): |
---|
| 701 | elem.preprocessed.add(self) |
---|
| 702 | return elem.state |
---|
| 703 | |
---|
| 704 | ret = False |
---|
| 705 | elements = [getobj(elem) for elem in |
---|
| 706 | self.targettask.filter_polymorphic_elements(self._mappers) |
---|
| 707 | if self not in elem.preprocessed and not elem.isdelete] |
---|
| 708 | if elements: |
---|
| 709 | ret = True |
---|
| 710 | self.processor.preprocess_dependencies(self.targettask, elements, trans, delete=False) |
---|
| 711 | |
---|
| 712 | elements = [getobj(elem) for elem in |
---|
| 713 | self.targettask.filter_polymorphic_elements(self._mappers) |
---|
| 714 | if self not in elem.preprocessed and elem.isdelete] |
---|
| 715 | if elements: |
---|
| 716 | ret = True |
---|
| 717 | self.processor.preprocess_dependencies(self.targettask, elements, trans, delete=True) |
---|
| 718 | return ret |
---|
| 719 | |
---|
| 720 | def execute(self, trans, delete): |
---|
| 721 | """process all objects contained within this ``UOWDependencyProcessor``s target task.""" |
---|
| 722 | |
---|
| 723 | |
---|
| 724 | elements = [e for e in |
---|
| 725 | self.targettask.filter_polymorphic_elements(self._mappers) |
---|
| 726 | if bool(e.isdelete)==delete] |
---|
| 727 | |
---|
| 728 | self.processor.process_dependencies( |
---|
| 729 | self.targettask, |
---|
| 730 | [elem.state for elem in elements], |
---|
| 731 | trans, |
---|
| 732 | delete=delete) |
---|
| 733 | |
---|
| 734 | def get_object_dependencies(self, state, trans, passive): |
---|
| 735 | return trans.get_attribute_history(state, self.processor.key, passive=passive) |
---|
| 736 | |
---|
| 737 | def whose_dependent_on_who(self, state1, state2): |
---|
| 738 | """establish which object is operationally dependent amongst a parent/child |
---|
| 739 | using the semantics stated by the dependency processor. |
---|
| 740 | |
---|
| 741 | This method is used to establish a partial ordering (set of dependency tuples) |
---|
| 742 | when toplogically sorting on a per-instance basis. |
---|
| 743 | |
---|
| 744 | """ |
---|
| 745 | return self.processor.whose_dependent_on_who(state1, state2) |
---|
| 746 | |
---|
| 747 | class UOWExecutor(object): |
---|
| 748 | """Encapsulates the execution traversal of a UOWTransaction structure.""" |
---|
| 749 | |
---|
| 750 | def execute(self, trans, tasks, isdelete=None): |
---|
| 751 | if isdelete is not True: |
---|
| 752 | for task in tasks: |
---|
| 753 | self.execute_save_steps(trans, task) |
---|
| 754 | if isdelete is not False: |
---|
| 755 | for task in reversed(tasks): |
---|
| 756 | self.execute_delete_steps(trans, task) |
---|
| 757 | |
---|
| 758 | def save_objects(self, trans, task): |
---|
| 759 | task.mapper._save_obj(task.polymorphic_tosave_objects, trans) |
---|
| 760 | |
---|
| 761 | def delete_objects(self, trans, task): |
---|
| 762 | task.mapper._delete_obj(task.polymorphic_todelete_objects, trans) |
---|
| 763 | |
---|
| 764 | def execute_dependency(self, trans, dep, isdelete): |
---|
| 765 | dep.execute(trans, isdelete) |
---|
| 766 | |
---|
| 767 | def execute_save_steps(self, trans, task): |
---|
| 768 | self.save_objects(trans, task) |
---|
| 769 | for dep in task.polymorphic_cyclical_dependencies: |
---|
| 770 | self.execute_dependency(trans, dep, False) |
---|
| 771 | for dep in task.polymorphic_cyclical_dependencies: |
---|
| 772 | self.execute_dependency(trans, dep, True) |
---|
| 773 | self.execute_cyclical_dependencies(trans, task, False) |
---|
| 774 | self.execute_dependencies(trans, task) |
---|
| 775 | |
---|
| 776 | def execute_delete_steps(self, trans, task): |
---|
| 777 | self.execute_cyclical_dependencies(trans, task, True) |
---|
| 778 | self.delete_objects(trans, task) |
---|
| 779 | |
---|
| 780 | def execute_dependencies(self, trans, task): |
---|
| 781 | polymorphic_dependencies = list(task.polymorphic_dependencies) |
---|
| 782 | for dep in polymorphic_dependencies: |
---|
| 783 | self.execute_dependency(trans, dep, False) |
---|
| 784 | for dep in reversed(polymorphic_dependencies): |
---|
| 785 | self.execute_dependency(trans, dep, True) |
---|
| 786 | |
---|
| 787 | def execute_cyclical_dependencies(self, trans, task, isdelete): |
---|
| 788 | for t in task.dependent_tasks: |
---|
| 789 | self.execute(trans, [t], isdelete) |
---|