unitofwork.py

SQLAlchemy. 经典的Python ORM框架。学习必看。

            except KeyError:
                dp = dependencies.setdefault(state, {})
            try:
                l = dp[depprocessor]
            except KeyError:
                l = UOWTask(self.uowtransaction, depprocessor.targettask.mapper)
                dp[depprocessor] = l
            return l

        def dependency_in_cycles(dep):
            proctask = trans.get_task_by_mapper(dep.processor.mapper.base_mapper, True)
            targettask = trans.get_task_by_mapper(dep.targettask.mapper.base_mapper, True)
            return targettask in cycles and (proctask is not None and proctask in cycles)

        # organize all original UOWDependencyProcessors by their target task
        deps_by_targettask = {}
        for task in cycles:
            for dep in task.polymorphic_dependencies:
                if not dependency_in_cycles(dep):
                    extradeplist.append(dep)
                for t in dep.targettask.polymorphic_tasks():
                    l = deps_by_targettask.setdefault(t, [])
                    l.append(dep)

        object_to_original_task = {}

        for task in cycles:
            for subtask in task.polymorphic_tasks():
                for taskelement in subtask.elements:
                    state = taskelement.state
                    object_to_original_task[state] = subtask
                    for dep in deps_by_targettask.get(subtask, []):
                        # is this dependency involved in one of the cycles ?
                        # (don't count the DetectKeySwitch prop)
                        if dep.processor.no_dependencies or not dependency_in_cycles(dep):
                            continue
                        (processor, targettask) = (dep.processor, dep.targettask)
                        isdelete = taskelement.isdelete

                        # list of dependent objects from this object
                        (added, unchanged, deleted) = dep.get_object_dependencies(state, trans, passive=True)
                        if not added and not unchanged and not deleted:
                            continue
                            
                        # the task corresponding to saving/deleting of those dependent objects
                        childtask = trans.get_task_by_mapper(processor.mapper)

                        childlist = added + unchanged + deleted

                        for o in childlist:
                            # other object is None.  this can occur if the relationship is many-to-one
                            # or one-to-one, and None was set.  the "removed" object will be picked
                            # up in this iteration via the deleted_items() part of the collection.
                            if o is None:
                                continue

                            # the other object is not in the UOWTransaction !  but if we are many-to-one,
                            # we need a task in order to attach dependency operations, so establish a "listonly"
                            # task
                            if o not in childtask:
                                childtask.append(o, listonly=True)
                                object_to_original_task[o] = childtask

                            # create a tuple representing the "parent/child"
                            whosdep = dep.whose_dependent_on_who(state, o)
                            if whosdep is not None:
                                # append the tuple to the partial ordering.
                                tuples.append(whosdep)

                                # create a UOWDependencyProcessor representing this pair of objects.
                                # append it to a UOWTask
                                if whosdep[0] is state:
                                    get_dependency_task(whosdep[0], dep).append(whosdep[0], isdelete=isdelete)
                                else:
                                    get_dependency_task(whosdep[0], dep).append(whosdep[1], isdelete=isdelete)
                            else:
                                # TODO: no test coverage here
                                get_dependency_task(state, dep).append(state, isdelete=isdelete)

        head = topological.sort_as_tree(tuples, allobjects)
        
        used_tasks = util.Set()
        def make_task_tree(node, parenttask, nexttasks):
            (state, cycles, children) = node
            originating_task = object_to_original_task[state]
            used_tasks.add(originating_task)
            t = nexttasks.get(originating_task, None)
            if t is None:
                t = UOWTask(self.uowtransaction, originating_task.mapper)
                nexttasks[originating_task] = t
                parenttask._append_cyclical_childtask(t)
            t.append(state, originating_task._objects[state].listonly, isdelete=originating_task._objects[state].isdelete)

            if state in dependencies:
                for depprocessor, deptask in dependencies[state].iteritems():
                    t.cyclical_dependencies.add(depprocessor.branch(deptask))
            nd = {}
            for n in children:
                t2 = make_task_tree(n, t, nd)
            return t

        t = UOWTask(self.uowtransaction, self.mapper)
        
        # stick the non-circular dependencies onto the new UOWTask
        for d in extradeplist:
            t.dependencies.add(d)
        
        if head is not None:
            make_task_tree(head, t, {})

        ret = [t]
        for t2 in cycles:
            if t2 not in used_tasks and t2 is not self:
                # add tasks that were in the cycle, but didnt get assembled
                # into the cyclical tree, to the start of the list
                # TODO: no test coverage for this !!
                localtask = UOWTask(self.uowtransaction, t2.mapper)
                for state in t2.elements:
                    localtask.append(state, t2.listonly, isdelete=t2._objects[state].isdelete)
                for dep in t2.dependencies:
                    localtask.dependencies.add(dep)
                ret.insert(0, localtask)
        
        return ret

    def __repr__(self):
        if self.mapper is not None:
            if self.mapper.__class__.__name__ == 'Mapper':
                name = self.mapper.class_.__name__ + "/" + self.mapper.local_table.description
            else:
                name = repr(self.mapper)
        else:
            name = '(none)'
        return ("UOWTask(%s) Mapper: '%s'" % (hex(id(self)), name))

class UOWTaskElement(object):
    """An element within a UOWTask.

    Corresponds to a single object instance to be saved, deleted, or
    just part of the transaction as a placeholder for further
    dependencies (i.e. 'listonly').

    may also store additional sub-UOWTasks.
    """

    def __init__(self, state):
        self.state = state
        self.listonly = True
        self.childtasks = []
        self.isdelete = False
        self.__preprocessed = {}

    def update(self, listonly, isdelete):
        if not listonly and self.listonly:
            self.listonly = False
            self.__preprocessed.clear()
        if isdelete and not self.isdelete:
            self.isdelete = True
            self.__preprocessed.clear()

    def mark_preprocessed(self, processor):
        """Mark this element as *preprocessed* by a particular ``UOWDependencyProcessor``.

        Preprocessing is used by dependency.py to apply
        flush-time cascade rules to relations and bring all
        required objects into the flush context.

        each processor as marked as "processed" when complete, however
        changes to the state of this UOWTaskElement will reset
        the list of completed processors, so that they 
        execute again, until no new objects or state changes
        are brought in.
        """

        self.__preprocessed[processor] = True

    def is_preprocessed(self, processor):
        return self.__preprocessed.get(processor, False)

    def __repr__(self):
        return "UOWTaskElement/%d: %s/%d %s" % (id(self), self.obj.__class__.__name__, id(self.obj), (self.listonly and 'listonly' or (self.isdelete and 'delete' or 'save')) )

class UOWDependencyProcessor(object):
    """In between the saving and deleting of objects, process
    *dependent* data, such as filling in a foreign key on a child item
    from a new primary key, or deleting association rows before a
    delete.  This object acts as a proxy to a DependencyProcessor.
    """

    def __init__(self, processor, targettask):
        self.processor = processor
        self.targettask = targettask

    def __repr__(self):
        return "UOWDependencyProcessor(%s, %s)" % (str(self.processor), str(self.targettask))
    
    def __str__(self):
        return repr(self)
            
    def __eq__(self, other):
        return other.processor is self.processor and other.targettask is self.targettask

    def __hash__(self):
        return hash((self.processor, self.targettask))

    def preexecute(self, trans):
        """preprocess all objects contained within this ``UOWDependencyProcessor``s target task.

        This may locate additional objects which should be part of the
        transaction, such as those affected deletes, orphans to be
        deleted, etc.
        
        Once an object is preprocessed, its ``UOWTaskElement`` is marked as processed.  If subsequent 
        changes occur to the ``UOWTaskElement``, its processed flag is reset, and will require processing
        again.

        Return True if any objects were preprocessed, or False if no
        objects were preprocessed.  If True is returned, the parent ``UOWTransaction`` will
        ultimately call ``preexecute()`` again on all processors until no new objects are processed.
        """

        def getobj(elem):
            elem.mark_preprocessed(self)
            return elem.state

        ret = False
        elements = [getobj(elem) for elem in self.targettask.polymorphic_tosave_elements if elem.state is not None and not elem.is_preprocessed(self)]
        if elements:
            ret = True
            self.processor.preprocess_dependencies(self.targettask, elements, trans, delete=False)

        elements = [getobj(elem) for elem in self.targettask.polymorphic_todelete_elements if elem.state is not None and not elem.is_preprocessed(self)]
        if elements:
            ret = True
            self.processor.preprocess_dependencies(self.targettask, elements, trans, delete=True)
        return ret

    def execute(self, trans, delete):
        """process all objects contained within this ``UOWDependencyProcessor``s target task."""
        
        if not delete:
            self.processor.process_dependencies(self.targettask, [elem.state for elem in self.targettask.polymorphic_tosave_elements if elem.state is not None], trans, delete=False)
        else:
            self.processor.process_dependencies(self.targettask, [elem.state for elem in self.targettask.polymorphic_todelete_elements if elem.state is not None], trans, delete=True)

    def get_object_dependencies(self, state, trans, passive):
        return trans.get_attribute_history(state, self.processor.key, passive=passive)

    def whose_dependent_on_who(self, state1, state2):
        """establish which object is operationally dependent amongst a parent/child 
        using the semantics stated by the dependency processor.
        
        This method is used to establish a partial ordering (set of dependency tuples)
        when toplogically sorting on a per-instance basis.
        
        """
        
        return self.processor.whose_dependent_on_who(state1, state2)

    def branch(self, task):
        """create a copy of this ``UOWDependencyProcessor`` against a new ``UOWTask`` object.
        
        this is used within the instance-level sorting operation when a single ``UOWTask``
        is broken up into many individual ``UOWTask`` objects.
        
        """
        
        return UOWDependencyProcessor(self.processor, task)
    
        
class UOWExecutor(object):
    """Encapsulates the execution traversal of a UOWTransaction structure."""

    def execute(self, trans, tasks, isdelete=None):
        if isdelete is not True:
            for task in tasks:
                self.execute_save_steps(trans, task)
        if isdelete is not False:
            for task in util.reversed(tasks):
                self.execute_delete_steps(trans, task)

    def save_objects(self, trans, task):
        task.mapper._save_obj(task.polymorphic_tosave_objects, trans)

    def delete_objects(self, trans, task):
        task.mapper._delete_obj(task.polymorphic_todelete_objects, trans)

    def execute_dependency(self, trans, dep, isdelete):
        dep.execute(trans, isdelete)

    def execute_save_steps(self, trans, task):
        self.save_objects(trans, task)
        self.execute_cyclical_dependencies(trans, task, False)
        self.execute_per_element_childtasks(trans, task, False)
        self.execute_dependencies(trans, task, False)
        self.execute_dependencies(trans, task, True)
        
    def execute_delete_steps(self, trans, task):
        self.execute_cyclical_dependencies(trans, task, True)
        self.execute_per_element_childtasks(trans, task, True)
        self.delete_objects(trans, task)

    def execute_dependencies(self, trans, task, isdelete=None):
        if isdelete is not True:
            for dep in task.polymorphic_dependencies:
                self.execute_dependency(trans, dep, False)
        if isdelete is not False:
            for dep in util.reversed(list(task.polymorphic_dependencies)):
                self.execute_dependency(trans, dep, True)

    def execute_cyclical_dependencies(self, trans, task, isdelete):
        for dep in task.polymorphic_cyclical_dependencies:
            self.execute_dependency(trans, dep, isdelete)

    def execute_per_element_childtasks(self, trans, task, isdelete):
        for element in task.polymorphic_tosave_elements + task.polymorphic_todelete_elements:
            self.execute_element_childtasks(trans, element, isdelete)

    def execute_element_childtasks(self, trans, element, isdelete):
        for child in element.childtasks:
            self.execute(trans, [child], isdelete)