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) |
---|