attributes.py

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

                self.dict.pop(key, None)
                self.committed_state.pop(key, None)
                self.expired_attributes.add(key)
                if getattr(self.class_, key).impl.accepts_scalar_loader:
                    self.callables[key] = self

    def reset(self, key):
        """remove the given attribute and any callables associated with it."""
        self.dict.pop(key, None)
        self.callables.pop(key, None)

    def commit_attr(self, attr, value):
        if hasattr(attr, 'commit_to_state'):
            attr.commit_to_state(self, value)
        else:
            self.committed_state.pop(attr.key, None)
        self.pending.pop(attr.key, None)
        self.appenders.pop(attr.key, None)

    def commit(self, keys):
        """commit all attributes named in the given list of key names.

        This is used by a partial-attribute load operation to mark committed those attributes
        which were refreshed from the database.
        """

        if self.class_._class_state.has_mutable_scalars:
            for key in keys:
                attr = getattr(self.class_, key).impl
                if hasattr(attr, 'commit_to_state') and attr.key in self.dict:
                    attr.commit_to_state(self, self.dict[attr.key])
                else:
                    self.committed_state.pop(attr.key, None)
                self.pending.pop(key, None)
                self.appenders.pop(key, None)
        else:
            for key in keys:
                self.committed_state.pop(key, None)
                self.pending.pop(key, None)
                self.appenders.pop(key, None)

    def commit_all(self):
        """commit all attributes unconditionally.

        This is used after a flush() or a regular instance load or refresh operation
        to mark committed all populated attributes.
        """

        self.committed_state = {}
        self.modified = False
        self.pending = {}
        self.appenders = {}

        if self.class_._class_state.has_mutable_scalars:
            for attr in _managed_attributes(self.class_):
                if hasattr(attr.impl, 'commit_to_state') and attr.impl.key in self.dict:
                    attr.impl.commit_to_state(self, self.dict[attr.impl.key])

        # remove strong ref
        self._strong_obj = None


class WeakInstanceDict(UserDict.UserDict):
    """similar to WeakValueDictionary, but wired towards 'state' objects."""

    def __init__(self, *args, **kw):
        self._wr = weakref.ref(self)
        # RLock because the mutex is used by a cleanup
        # handler, which can be called at any time (including within an already mutexed block)
        self._mutex = threading.RLock()
        UserDict.UserDict.__init__(self, *args, **kw)

    def __getitem__(self, key):
        state = self.data[key]
        o = state.obj()
        if o is None:
            o = state._check_resurrect(self)
        if o is None:
            raise KeyError, key
        return o

    def __contains__(self, key):
        try:
            state = self.data[key]
            o = state.obj()
            if o is None:
                o = state._check_resurrect(self)
        except KeyError:
            return False
        return o is not None

    def has_key(self, key):
        return key in self

    def __repr__(self):
        return "<InstanceDict at %s>" % id(self)

    def __setitem__(self, key, value):
        if key in self.data:
            self._mutex.acquire()
            try:
                if key in self.data:
                    self.data[key].instance_dict = None
            finally:
                self._mutex.release()
        self.data[key] = value._state
        value._state.instance_dict = self._wr

    def __delitem__(self, key):
        state = self.data[key]
        state.instance_dict = None
        del self.data[key]

    def get(self, key, default=None):
        try:
            state = self.data[key]
        except KeyError:
            return default
        else:
            o = state.obj()
            if o is None:
                # This should only happen
                return default
            else:
                return o

    def items(self):
        L = []
        for key, state in self.data.items():
            o = state.obj()
            if o is not None:
                L.append((key, o))
        return L

    def iteritems(self):
        for state in self.data.itervalues():
            value = state.obj()
            if value is not None:
                yield value._instance_key, value

    def iterkeys(self):
        return self.data.iterkeys()

    def __iter__(self):
        return self.data.iterkeys()

    def __len__(self):
        return len(self.values())

    def itervalues(self):
        for state in self.data.itervalues():
            instance = state.obj()
            if instance is not None:
                yield instance

    def values(self):
        L = []
        for state in self.data.values():
            o = state.obj()
            if o is not None:
                L.append(o)
        return L

    def popitem(self):
        raise NotImplementedError()

    def pop(self, key, *args):
        raise NotImplementedError()

    def setdefault(self, key, default=None):
        raise NotImplementedError()

    def update(self, dict=None, **kwargs):
        raise NotImplementedError()

    def copy(self):
        raise NotImplementedError()

    def all_states(self):
        return self.data.values()

class StrongInstanceDict(dict):
    def all_states(self):
        return [o._state for o in self.values()]

def _create_history(attr, state, current):
    original = state.committed_state.get(attr.key, NEVER_SET)

    if hasattr(attr, 'get_collection'):
        current = attr.get_collection(state, current)
        if original is NO_VALUE:
            return (list(current), [], [])
        elif original is NEVER_SET:
            return ([], list(current), [])
        else:
            collection = util.OrderedIdentitySet(current)
            s = util.OrderedIdentitySet(original)
            return (list(collection.difference(s)), list(collection.intersection(s)), list(s.difference(collection)))
    else:
        if current is NO_VALUE:
            if original not in [None, NEVER_SET, NO_VALUE]:
                deleted = [original]
            else:
                deleted = []
            return ([], [], deleted)
        elif original is NO_VALUE:
            return ([current], [], [])
        elif original is NEVER_SET or attr.is_equal(current, original) is True:   # dont let ClauseElement expressions here trip things up
            return ([], [current], [])
        else:
            if original is not None:
                deleted = [original]
            else:
                deleted = []
            return ([current], [], deleted)

class PendingCollection(object):
    """stores items appended and removed from a collection that has not been loaded yet.

    When the collection is loaded, the changes present in PendingCollection are applied
    to produce the final result.
    """

    def __init__(self):
        self.deleted_items = util.IdentitySet()
        self.added_items = util.OrderedIdentitySet()

    def append(self, value):
        if value in self.deleted_items:
            self.deleted_items.remove(value)
        self.added_items.add(value)

    def remove(self, value):
        if value in self.added_items:
            self.added_items.remove(value)
        self.deleted_items.add(value)

def _managed_attributes(class_):
    """return all InstrumentedAttributes associated with the given class_ and its superclasses."""

    return chain(*[cl._class_state.attrs.values() for cl in class_.__mro__[:-1] if hasattr(cl, '_class_state')])

def get_history(state, key, **kwargs):
    return getattr(state.class_, key).impl.get_history(state, **kwargs)

def get_as_list(state, key, passive=False):
    """return an InstanceState attribute as a list,
    regardless of it being a scalar or collection-based
    attribute.

    returns None if passive=True and the getter returns
    PASSIVE_NORESULT.
    """

    attr = getattr(state.class_, key).impl
    x = attr.get(state, passive=passive)
    if x is PASSIVE_NORESULT:
        return None
    elif hasattr(attr, 'get_collection'):
        return attr.get_collection(state, x)
    elif isinstance(x, list):
        return x
    else:
        return [x]

def has_parent(class_, instance, key, optimistic=False):
    return getattr(class_, key).impl.hasparent(instance._state, optimistic=optimistic)

def _create_prop(class_, key, uselist, callable_, typecallable, useobject, mutable_scalars, **kwargs):
    if kwargs.pop('dynamic', False):
        from sqlalchemy.orm import dynamic
        return dynamic.DynamicAttributeImpl(class_, key, typecallable, **kwargs)
    elif uselist:
        return CollectionAttributeImpl(class_, key, callable_, typecallable, **kwargs)
    elif useobject:
        return ScalarObjectAttributeImpl(class_, key, callable_,**kwargs)
    elif mutable_scalars:
        return MutableScalarAttributeImpl(class_, key, callable_, **kwargs)
    else:
        return ScalarAttributeImpl(class_, key, callable_, **kwargs)

def manage(instance):
    """initialize an InstanceState on the given instance."""

    if not hasattr(instance, '_state'):
        instance._state = InstanceState(instance)

def new_instance(class_, state=None):
    """create a new instance of class_ without its __init__() method being called.

    Also initializes an InstanceState on the new instance.
    """

    s = class_.__new__(class_)
    if state:
        s._state = state
    else:
        s._state = InstanceState(s)
    return s

def _init_class_state(class_):
    if not '_class_state' in class_.__dict__:
        class_._class_state = ClassState()

def register_class(class_, extra_init=None, on_exception=None, deferred_scalar_loader=None):
    # do a sweep first, this also helps some attribute extensions
    # (like associationproxy) become aware of themselves at the
    # class level
    for key in dir(class_):
        getattr(class_, key, None)

    _init_class_state(class_)
    class_._class_state.deferred_scalar_loader=deferred_scalar_loader

    oldinit = None
    doinit = False

    def init(instance, *args, **kwargs):
        if not hasattr(instance, '_state'):
            instance._state = InstanceState(instance)

        if extra_init:
            extra_init(class_, oldinit, instance, args, kwargs)

        try:
            if doinit:
                oldinit(instance, *args, **kwargs)
            elif args or kwargs:
                # simulate error message raised by object(), but don't copy
                # the text verbatim
                raise TypeError("default constructor for object() takes no parameters")
        except:
            if on_exception:
                on_exception(class_, oldinit, instance, args, kwargs)
            raise


    # override oldinit
    oldinit = class_.__init__
    if oldinit is None or not hasattr(oldinit, '_oldinit'):
        init._oldinit = oldinit
        class_.__init__ = init
    # if oldinit is already one of our 'init' methods, replace it
    elif hasattr(oldinit, '_oldinit'):
        init._oldinit = oldinit._oldinit
        class_.__init = init
        oldinit = oldinit._oldinit

    if oldinit is not None:
        doinit = oldinit is not object.__init__
        try:
            init.__name__ = oldinit.__name__
            init.__doc__ = oldinit.__doc__
        except:
            # cant set __name__ in py 2.3 !
            pass

def unregister_class(class_):
    if hasattr(class_, '__init__') and hasattr(class_.__init__, '_oldinit'):
        if class_.__init__._oldinit is not None:
            class_.__init__ = class_.__init__._oldinit
        else:
            delattr(class_, '__init__')

    if '_class_state' in class_.__dict__:
        _class_state = class_.__dict__['_class_state']
        for key, attr in _class_state.attrs.iteritems():
            if key in class_.__dict__:
                delattr(class_, attr.impl.key)
        delattr(class_, '_class_state')

def register_attribute(class_, key, uselist, useobject, callable_=None, proxy_property=None, mutable_scalars=False, **kwargs):
    _init_class_state(class_)

    typecallable = kwargs.pop('typecallable', None)
    if isinstance(typecallable, InstrumentedAttribute):
        typecallable = None
    comparator = kwargs.pop('comparator', None)

    if key in class_.__dict__ and isinstance(class_.__dict__[key], InstrumentedAttribute):
        # this currently only occurs if two primary mappers are made for the same class.
        # TODO:  possibly have InstrumentedAttribute check "entity_name" when searching for impl.
        # raise an error if two attrs attached simultaneously otherwise
        return

    if proxy_property:
        inst = ProxiedAttribute(key, proxy_property, comparator=comparator)
    else:
        inst = InstrumentedAttribute(_create_prop(class_, key, uselist, callable_, useobject=useobject,
                                       typecallable=typecallable, mutable_scalars=mutable_scalars, **kwargs), comparator=comparator)

    setattr(class_, key, inst)
    class_._class_state.attrs[key] = inst

def unregister_attribute(class_, key):
    class_state = class_._class_state
    if key in class_state.attrs:
        del class_._class_state.attrs[key]
        delattr(class_, key)

def init_collection(instance, key):
    """Initialize a collection attribute and return the collection adapter."""
    attr = getattr(instance.__class__, key).impl
    state = instance._state
    user_data = attr.initialize(state)
    return attr.get_collection(state, user_data)