unitofwork.py

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

        this indicates that an INSERT statement elsewhere corresponds to this DELETE;
        the INSERT is converted to an UPDATE and the DELETE does not occur.
        """
        mapper = _state_mapper(state)
        task = self.get_task_by_mapper(mapper)
        taskelement = task._objects[state]
        taskelement.isdelete = "rowswitch"
        
    def unregister_object(self, obj):
        """remove an object from its parent UOWTask.
        
        called by mapper._save_obj() when an 'identity switch' is detected, so that
        no further operations occur upon the instance."""
        mapper = object_mapper(obj)
        task = self.get_task_by_mapper(mapper)
        if obj._state in task._objects:
            task.delete(obj._state)

    def is_deleted(self, state):
        """return true if the given state is marked as deleted within this UOWTransaction."""
        
        mapper = _state_mapper(state)
        task = self.get_task_by_mapper(mapper)
        return task.is_deleted(state)
        
    def get_task_by_mapper(self, mapper, dontcreate=False):
        """return UOWTask element corresponding to the given mapper.

        Will create a new UOWTask, including a UOWTask corresponding to the 
        "base" inherited mapper, if needed, unless the dontcreate flag is True.
        """
        try:
            return self.tasks[mapper]
        except KeyError:
            if dontcreate:
                return None
                
            base_mapper = mapper.base_mapper
            if base_mapper in self.tasks:
                base_task = self.tasks[base_mapper]
            else:
                self.tasks[base_mapper] = base_task = UOWTask(self, base_mapper)
                base_mapper._register_dependencies(self)

            if mapper not in self.tasks:
                self.tasks[mapper] = task = UOWTask(self, mapper, base_task=base_task)
                mapper._register_dependencies(self)
            else:
                task = self.tasks[mapper]
                
            return task

    def register_dependency(self, mapper, dependency):
        """register a dependency between two mappers.

        Called by ``mapper.PropertyLoader`` to register the objects
        handled by one mapper being dependent on the objects handled
        by another.        
        """

        # correct for primary mapper
        # also convert to the "base mapper", the parentmost task at the top of an inheritance chain
        # dependency sorting is done via non-inheriting mappers only, dependencies between mappers
        # in the same inheritance chain is done at the per-object level
        mapper = mapper.primary_mapper().base_mapper
        dependency = dependency.primary_mapper().base_mapper

        self.dependencies.add((mapper, dependency))

    def register_processor(self, mapper, processor, mapperfrom):
        """register a dependency processor, corresponding to dependencies between
        the two given mappers.
        
        """

        # correct for primary mapper
        mapper = mapper.primary_mapper()
        mapperfrom = mapperfrom.primary_mapper()

        task = self.get_task_by_mapper(mapper)
        targettask = self.get_task_by_mapper(mapperfrom)
        up = UOWDependencyProcessor(processor, targettask)
        task.dependencies.add(up)
        
    def execute(self):
        """Execute this UOWTransaction.
        
        This will organize all collected UOWTasks into a dependency-sorted
        list which is then traversed using the traversal scheme
        encoded in the UOWExecutor class.  Operations to mappers and dependency
        processors are fired off in order to issue SQL to the database and 
        synchronize instance attributes with database values and related
        foreign key values."""

        # pre-execute dependency processors.  this process may
        # result in new tasks, objects and/or dependency processors being added,
        # particularly with 'delete-orphan' cascade rules.
        # keep running through the full list of tasks until all
        # objects have been processed.
        while True:
            ret = False
            for task in self.tasks.values():
                for up in list(task.dependencies):
                    if up.preexecute(self):
                        ret = True
            if not ret:
                break

        tasks = self._sort_dependencies()
        if self._should_log_info:
            self.logger.info("Task dump:\n" + self._dump(tasks))
        UOWExecutor().execute(self, tasks)
        if self._should_log_info:
            self.logger.info("Execute Complete")

    def _dump(self, tasks):
        buf = StringIO.StringIO()
        import uowdumper
        uowdumper.UOWDumper(tasks, buf)
        return buf.getvalue()
        
    def post_exec(self):
        """mark processed objects as clean / deleted after a successful flush().
        
        this method is called within the flush() method after the 
        execute() method has succeeded and the transaction has been committed.
        """

        for task in self.tasks.values():
            for elem in task.elements:
                if elem.state is None:
                    continue
                if elem.isdelete:
                    self.uow._remove_deleted(elem.state)
                else:
                    self.uow._register_clean(elem.state)

    def _sort_dependencies(self):
        nodes = topological.sort_with_cycles(self.dependencies, 
            [t.mapper for t in self.tasks.values() if t.base_task is t]
        )

        ret = []
        for item, cycles in nodes:
            task = self.get_task_by_mapper(item)
            if cycles:
                for t in task._sort_circular_dependencies(self, [self.get_task_by_mapper(i) for i in cycles]):
                    ret.append(t)
            else:
                ret.append(task)

        if self._should_log_debug:
            self.logger.debug("Dependent tuples:\n" + "\n".join(["(%s->%s)" % (d[0].class_.__name__, d[1].class_.__name__) for d in self.dependencies]))
            self.logger.debug("Dependency sort:\n"+ str(ret))
        return ret

class UOWTask(object):
    """Represents all of the objects in the UOWTransaction which correspond to
    a particular mapper.  This is the primary class of three classes used to generate
    the elements of the dependency graph.
    """

    def __init__(self, uowtransaction, mapper, base_task=None):
        self.uowtransaction = uowtransaction

        # base_task is the UOWTask which represents the "base mapper"
        # in our mapper's inheritance chain.  if the mapper does not
        # inherit from any other mapper, the base_task is self.
        # the _inheriting_tasks dictionary is a dictionary present only
        # on the "base_task"-holding UOWTask, which maps all mappers within
        # an inheritance hierarchy to their corresponding UOWTask instances.
        if base_task is None:
            self.base_task = self
            self._inheriting_tasks = {mapper:self}
        else:
            self.base_task = base_task
            base_task._inheriting_tasks[mapper] = self
        
        # the Mapper which this UOWTask corresponds to
        self.mapper = mapper

        # mapping of InstanceState -> UOWTaskElement
        self._objects = {} 

        self.dependencies = util.Set()
        self.cyclical_dependencies = util.Set()

    def polymorphic_tasks(self):
        """return an iterator of UOWTask objects corresponding to the inheritance sequence
        of this UOWTask's mapper.
        
            e.g. if mapper B and mapper C inherit from mapper A, and mapper D inherits from B:
            
                mapperA -> mapperB -> mapperD
                        -> mapperC 
                                   
            the inheritance sequence starting at mapper A is a depth-first traversal:
            
                [mapperA, mapperB, mapperD, mapperC]
                
            this method will therefore return
            
                [UOWTask(mapperA), UOWTask(mapperB), UOWTask(mapperD), UOWTask(mapperC)]
                
        The concept of "polymporphic iteration" is adapted into several property-based 
        iterators which return object instances, UOWTaskElements and UOWDependencyProcessors
        in an order corresponding to this sequence of parent UOWTasks.  This is used to issue
        operations related to inheritance-chains of mappers in the proper order based on 
        dependencies between those mappers.
        
        """
        
        for mapper in self.mapper.polymorphic_iterator():
            t = self.base_task._inheriting_tasks.get(mapper, None)
            if t is not None:
                yield t

    def is_empty(self):
        """return True if this UOWTask is 'empty', meaning it has no child items.

        used only for debugging output.
        """

        return not self._objects and not self.dependencies
            
    def append(self, state, listonly=False, isdelete=False):
        if state not in self._objects:
            self._objects[state] = rec = UOWTaskElement(state)
        else:
            rec = self._objects[state]
        
        rec.update(listonly, isdelete)
    
    def _append_cyclical_childtask(self, task):
        if "cyclical" not in self._objects:
            self._objects["cyclical"] = UOWTaskElement(None)
        self._objects["cyclical"].childtasks.append(task)

    def append_postupdate(self, state, post_update_cols):
        """issue a 'post update' UPDATE statement via this object's mapper immediately.  
        
        this operation is used only with relations that specify the `post_update=True`
        flag.
        """

        # postupdates are UPDATED immeditely (for now)
        # convert post_update_cols list to a Set so that __hashcode__ is used to compare columns
        # instead of __eq__
        self.mapper._save_obj([state], self.uowtransaction, postupdate=True, post_update_cols=util.Set(post_update_cols))

    def delete(self, obj):
        """remove the given object from this UOWTask, if present."""

        self._objects.pop(obj._state, None)

    def __contains__(self, state):
        """return True if the given object is contained within this UOWTask or inheriting tasks."""
        
        for task in self.polymorphic_tasks():
            if state in task._objects:
                return True
        else:
            return False

    def is_deleted(self, state):
        """return True if the given object is marked as to be deleted within this UOWTask."""
        
        try:
            return self._objects[state].isdelete
        except KeyError:
            return False

    def _polymorphic_collection(callable):
        """return a property that will adapt the collection returned by the
        given callable into a polymorphic traversal."""
        
        def collection(self):
            for task in self.polymorphic_tasks():
                for rec in callable(task):
                    yield rec
        return property(collection)
        
    elements = property(lambda self:self._objects.values())
    
    polymorphic_elements = _polymorphic_collection(lambda task:task.elements)

    polymorphic_tosave_elements = property(lambda self: [rec for rec in self.polymorphic_elements
                                             if not rec.isdelete])
                                             
    polymorphic_todelete_elements = property(lambda self:[rec for rec in self.polymorphic_elements
                                               if rec.isdelete])

    polymorphic_tosave_objects = property(lambda self:[rec.state for rec in self.polymorphic_elements
                                          if rec.state is not None and not rec.listonly and rec.isdelete is False])

    polymorphic_todelete_objects = property(lambda self:[rec.state for rec in self.polymorphic_elements
                                          if rec.state is not None and not rec.listonly and rec.isdelete is True])

    polymorphic_dependencies = _polymorphic_collection(lambda task:task.dependencies)
    
    polymorphic_cyclical_dependencies = _polymorphic_collection(lambda task:task.cyclical_dependencies)
    
    def _sort_circular_dependencies(self, trans, cycles):
        """Create a hierarchical tree of *subtasks*
        which associate specific dependency actions with individual
        objects.  This is used for a *cyclical* task, or a task where
        elements of its object list contain dependencies on each
        other.

        This is not the normal case; this logic only kicks in when
        something like a hierarchical tree is being represented.
        """
        allobjects = []
        for task in cycles:
            allobjects += [e.state for e in task.polymorphic_elements]
        tuples = []

        cycles = util.Set(cycles)

        extradeplist = []
        dependencies = {}

        def get_dependency_task(state, depprocessor):
            try:
                dp = dependencies[state]