def _register_clean(self, state):
        """register the given object as 'clean' (i.e. persistent) within this unit of work, after
        a save operation has taken place."""

        mapper = _state_mapper(state)
        instance_key = mapper._identity_key_from_state(state)
        
        if '_instance_key' not in state.dict:
            state.dict['_instance_key'] = instance_key
            
        elif state.dict['_instance_key'] != instance_key:
            # primary key switch
            del self.identity_map[state.dict['_instance_key']]
            state.dict['_instance_key'] = instance_key
            
        if hasattr(state, 'insert_order'):
            delattr(state, 'insert_order')
        
        o = state.obj()
        # prevent against last minute dereferences of the object
        # TODO: identify a code path where state.obj() is None
        if o is not None:
            self.identity_map[state.dict['_instance_key']] = o
            state.commit_all()
        
        # remove from new last, might be the last strong ref
        self.new.pop(state, None)

    def register_object(self, state, isdelete=False, listonly=False, postupdate=False, post_update_cols=None):

        # if object is not in the overall session, do nothing
        if not self.session._contains_state(state):
            if self._should_log_debug:
                self.logger.debug("object %s not part of session, not registering for flush" % (mapperutil.state_str(state)))
            return

        if self._should_log_debug:
            self.logger.debug("register object for flush: %s isdelete=%s listonly=%s postupdate=%s" % (mapperutil.state_str(state), isdelete, listonly, postupdate))

        mapper = _state_mapper(state)

        task = self.get_task_by_mapper(mapper)
        if postupdate:
            task.append_postupdate(state, post_update_cols)
        else:
            task.append(state, listonly=listonly, isdelete=isdelete)

        # ensure the mapper for this object has had its
        # DependencyProcessors added.
        if mapper not in self.processors:
            mapper._register_processors(self)
            self.processors.add(mapper)

            if mapper.base_mapper not in self.processors:
                mapper.base_mapper._register_processors(self)
                self.processors.add(mapper.base_mapper)

 def append(self, state, item, initiator):
     # process "save_update" cascade rules for when an instance is appended to the list of another instance
     sess = _state_session(state)
     if sess:
         prop = _state_mapper(state).get_property(self.key)
         if prop.cascade.save_update and item not in sess:
             sess.add(item)
     return item

 def remove(self, state, item, initiator):
     sess = _state_session(state)
     if sess:
         prop = _state_mapper(state).get_property(self.key)
         # expunge pending orphans
         if prop.cascade.delete_orphan and \
             item in sess.new and \
             prop.mapper._is_orphan(attributes.instance_state(item)):
                 sess.expunge(item)

 def set_row_switch(self, state):
     """mark a deleted object as a 'row switch'.
     
     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 __call__(self):
        state = self.state
        if not mapper._state_has_identity(state):
            return None

        instance_mapper = mapper._state_mapper(state)
        prop = instance_mapper.get_property(self.key)
        strategy = prop._get_strategy(LazyLoader)
        
        if strategy._should_log_debug:
            strategy.logger.debug("loading %s" % mapperutil.state_attribute_str(state, self.key))

        session = sessionlib._state_session(state)
        if session is None:
            raise sa_exc.UnboundExecutionError(
                "Parent instance %s is not bound to a Session; "
                "lazy load operation of attribute '%s' cannot proceed" % 
                (mapperutil.state_str(state), self.key)
            )
        
        q = session.query(prop.mapper)._adapt_all_clauses()
        
        if self.path:
            q = q._with_current_path(self.path)
            
        # if we have a simple primary key load, use mapper.get()
        # to possibly save a DB round trip
        if strategy.use_get:
            ident = []
            allnulls = True
            for primary_key in prop.mapper.primary_key: 
                val = instance_mapper._get_committed_state_attr_by_column(state, strategy._equated_columns[primary_key])
                allnulls = allnulls and val is None
                ident.append(val)
            if allnulls:
                return None
            if self.options:
                q = q._conditional_options(*self.options)
            return q.get(ident)

        if prop.order_by:
            q = q.order_by(*util.to_list(prop.order_by))

        if self.options:
            q = q._conditional_options(*self.options)
        q = q.filter(strategy.lazy_clause(state))

        result = q.all()
        if strategy.uselist:
            return result
        else:
            if result:
                return result[0]
            else:
                return None

 def set(self, state, newvalue, oldvalue, initiator):
     # process "save_update" cascade rules for when an instance is attached to another instance
     if oldvalue is newvalue:
         return newvalue
     sess = _state_session(state)
     if sess:
         prop = _state_mapper(state).get_property(self.key)
         if newvalue is not None and prop.cascade.save_update and newvalue not in sess:
             sess.add(newvalue)
         if prop.cascade.delete_orphan and oldvalue in sess.new:
             sess.expunge(oldvalue)
     return newvalue

    def class_level_loader(self, state, options=None, path=None):
        if not mapperutil._state_has_identity(state):
            return None

        localparent = mapper._state_mapper(state)

        # adjust for the PropertyLoader associated with the instance
        # not being our own PropertyLoader.
        # TODO: this may no longer be relevant without entity_name
        prop = localparent.get_property(self.key)
        if prop is not self.parent_property:
            return prop._get_strategy(LazyLoader).setup_loader(state)

        return LoadLazyAttribute(state, self.key, options, path)

    def register_object(self, state, isdelete=False, listonly=False, postupdate=False, post_update_cols=None):
        # if object is not in the overall session, do nothing
        if not self.session._contains_state(state):
            if self._should_log_debug:
                self.logger.debug("object %s not part of session, not registering for flush" % (mapperutil.state_str(state)))
            return

        if self._should_log_debug:
            self.logger.debug("register object for flush: %s isdelete=%s listonly=%s postupdate=%s" % (mapperutil.state_str(state), isdelete, listonly, postupdate))

        mapper = _state_mapper(state)

        task = self.get_task_by_mapper(mapper)
        if postupdate:
            task.append_postupdate(state, post_update_cols)
        else:
            task.append(state, listonly=listonly, isdelete=isdelete)

    def __call__(self, **kw):
        if kw.get('passive') is attributes.PASSIVE_NO_FETCH:
            return attributes.PASSIVE_NO_RESULT

        state = self.state
        
        localparent = mapper._state_mapper(state)
        
        prop = localparent.get_property(self.key)
        strategy = prop._get_strategy(DeferredColumnLoader)

        if strategy.group:
            toload = [
                    p.key for p in 
                    localparent.iterate_properties 
                    if isinstance(p, StrategizedProperty) and 
                      isinstance(p.strategy, DeferredColumnLoader) and 
                      p.group==strategy.group
                    ]
        else:
            toload = [self.key]

        # narrow the keys down to just those which have no history
        group = [k for k in toload if k in state.unmodified]

        if strategy._should_log_debug():
            strategy.logger.debug(
                    "deferred load %s group %s",
                    (mapperutil.state_attribute_str(state, self.key), 
                    group and ','.join(group) or 'None')
            )

        session = sessionlib._state_session(state)
        if session is None:
            raise orm_exc.DetachedInstanceError(
                "Parent instance %s is not bound to a Session; "
                "deferred load operation of attribute '%s' cannot proceed" % 
                (mapperutil.state_str(state), self.key)
                )

        query = session.query(localparent)
        ident = state.key[1]
        query._get(None, ident=ident, 
                    only_load_props=group, refresh_state=state)
        return attributes.ATTR_WAS_SET

    def __call__(self):
        state = self.state

        if not mapper._state_has_identity(state):
            return None

        localparent = mapper._state_mapper(state)

        prop = localparent.get_property(self.key)
        strategy = prop._get_strategy(DeferredColumnLoader)

        if self.keys:
            toload = self.keys
        elif strategy.group:
            toload = [
                p.key
                for p in localparent.iterate_properties
                if isinstance(p, StrategizedProperty)
                and isinstance(p.strategy, DeferredColumnLoader)
                and p.group == strategy.group
            ]
        else:
            toload = [self.key]

        # narrow the keys down to just those which have no history
        group = [k for k in toload if k in state.unmodified]

        if strategy._should_log_debug:
            strategy.logger.debug(
                "deferred load %s group %s"
                % (mapperutil.state_attribute_str(state, self.key), group and ",".join(group) or "None")
            )

        session = sessionlib._state_session(state)
        if session is None:
            raise sa_exc.UnboundExecutionError(
                "Parent instance %s is not bound to a Session; deferred load operation of attribute '%s' cannot proceed"
                % (mapperutil.state_str(state), self.key)
            )

        query = session.query(localparent)
        ident = state.key[1]
        query._get(None, ident=ident, only_load_props=group, refresh_state=state)
        return attributes.ATTR_WAS_SET

    def register_object(self, state, isdelete=False, 
                            listonly=False, postupdate=False, post_update_cols=None):
        
        # if object is not in the overall session, do nothing
        if not self.session._contains_state(state):
            return

        mapper = _state_mapper(state)

        task = self.get_task_by_mapper(mapper)
        if postupdate:
            task.append_postupdate(state, post_update_cols)
        else:
            task.append(state, listonly=listonly, isdelete=isdelete)

        # ensure the mapper for this object has had its 
        # DependencyProcessors added.
        if mapper not in self.processors:
            mapper._register_processors(self)
            self.processors.add(mapper)

            if mapper.base_mapper not in self.processors:
                mapper.base_mapper._register_processors(self)
                self.processors.add(mapper.base_mapper)

    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 __call__(self, **kw):
        state = self.state
        instance_mapper = mapper._state_mapper(state)
        prop = instance_mapper.get_property(self.key)
        strategy = prop._get_strategy(LazyLoader)

        if kw.get("passive") is attributes.PASSIVE_NO_FETCH and not strategy.use_get:
            return attributes.PASSIVE_NO_RESULT

        if strategy._should_log_debug():
            strategy.logger.debug("loading %s", mapperutil.state_attribute_str(state, self.key))

        session = sessionlib._state_session(state)
        if session is None:
            raise orm_exc.DetachedInstanceError(
                "Parent instance %s is not bound to a Session; "
                "lazy load operation of attribute '%s' cannot proceed" % (mapperutil.state_str(state), self.key)
            )

        q = session.query(prop.mapper)._adapt_all_clauses()

        if state.load_path:
            q = q._with_current_path(state.load_path + (self.key,))

        # if we have a simple primary key load, use mapper.get()
        # to possibly save a DB round trip
        if strategy.use_get:
            ident = []
            allnulls = True
            for primary_key in prop.mapper.primary_key:
                val = instance_mapper._get_committed_state_attr_by_column(
                    state, state.dict, strategy._equated_columns[primary_key], **kw
                )
                if val is attributes.PASSIVE_NO_RESULT:
                    return val
                allnulls = allnulls and val is None
                ident.append(val)

            if allnulls:
                return None

            if state.load_options:
                q = q._conditional_options(*state.load_options)

            key = prop.mapper.identity_key_from_primary_key(ident)
            return q._get(key, ident, **kw)

        if prop.order_by:
            q = q.order_by(*util.to_list(prop.order_by))

        for rev in prop._reverse_property:
            # reverse props that are MANYTOONE are loading *this*
            # object from get(), so don't need to eager out to those.
            if rev.direction is interfaces.MANYTOONE and rev._use_get and not isinstance(rev.strategy, LazyLoader):
                q = q.options(EagerLazyOption(rev.key, lazy="select"))

        if state.load_options:
            q = q._conditional_options(*state.load_options)

        q = q.filter(strategy.lazy_clause(state))

        result = q.all()
        if strategy.uselist:
            return result
        else:
            l = len(result)
            if l:
                if l > 1:
                    util.warn("Multiple rows returned with " "uselist=False for lazily-loaded attribute '%s' " % prop)

                return result[0]
            else:
                return None

    def flush(self, session, objects=None):
        """create a dependency tree of all pending SQL operations within this unit of work and execute."""

        dirty = [x for x in self.identity_map.all_states()
            if x.modified
            or (x.class_._class_state.has_mutable_scalars and x.is_modified())
        ]
        
        if not dirty and not self.deleted and not self.new:
            return
        
        deleted = util.Set(self.deleted)
        new = util.Set(self.new)
        
        dirty = util.Set(dirty).difference(deleted)
        
        flush_context = UOWTransaction(self, session)

        if session.extension is not None:
            session.extension.before_flush(session, flush_context, objects)

        # create the set of all objects we want to operate upon
        if objects:
            # specific list passed in
            objset = util.Set([o._state for o in objects])
        else:
            # or just everything
            objset = util.Set(self.identity_map.all_states()).union(new)
            
        # store objects whose fate has been decided
        processed = util.Set()

        # put all saves/updates into the flush context.  detect top-level orphans and throw them into deleted.
        for state in new.union(dirty).intersection(objset).difference(deleted):
            if state in processed:
                continue

            obj = state.obj()
            is_orphan = _state_mapper(state)._is_orphan(obj)
            if is_orphan and not has_identity(obj):
                raise exceptions.FlushError("instance %s is an unsaved, pending instance and is an orphan (is not attached to %s)" %
                    (
                        obj,
                        ", nor ".join(["any parent '%s' instance via that classes' '%s' attribute" % (klass.__name__, key) for (key,klass) in _state_mapper(state).delete_orphans])
                    ))
            flush_context.register_object(state, isdelete=is_orphan)
            processed.add(state)

        # put all remaining deletes into the flush context.
        for state in deleted.intersection(objset).difference(processed):
            flush_context.register_object(state, isdelete=True)

        if len(flush_context.tasks) == 0:
            return
            
        session.create_transaction(autoflush=False)
        flush_context.transaction = session.transaction
        try:
            flush_context.execute()
            
            if session.extension is not None:
                session.extension.after_flush(session, flush_context)
            session.commit()
        except:
            session.rollback()
            raise

        for state in deleted.intersection(objset):
            state.is_deleted = True

        flush_context.post_exec()

        if session.extension is not None:
            session.extension.after_flush_postexec(session, flush_context)