associationproxy.py

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

    def __add__(self, iterable):
        try:
            other = list(iterable)
        except TypeError:
            return NotImplemented
        return list(self) + other

    def __radd__(self, iterable):
        try:
            other = list(iterable)
        except TypeError:
            return NotImplemented
        return other + list(self)

    def __mul__(self, n):
        if not isinstance(n, int):
            return NotImplemented
        return list(self) * n
    __rmul__ = __mul__

    def __iadd__(self, iterable):
        self.extend(iterable)
        return self

    def __imul__(self, n):
        # unlike a regular list *=, proxied __imul__ will generate unique
        # backing objects for each copy.  *= on proxied lists is a bit of
        # a stretch anyhow, and this interpretation of the __imul__ contract
        # is more plausibly useful than copying the backing objects.
        if not isinstance(n, int):
            return NotImplemented
        if n == 0:
            self.clear()
        elif n > 1:
            self.extend(list(self) * (n - 1))
        return self

    def copy(self):
        return list(self)

    def __repr__(self):
        return repr(list(self))

    def __hash__(self):
        raise TypeError("%s objects are unhashable" % type(self).__name__)

_NotProvided = object()
class _AssociationDict(object):
    """Generic proxying list which proxies dict operations to a another dict,
    converting association objects to and from a simplified value.
    """

    def __init__(self, lazy_collection, creator, getter, setter):
        """
        lazy_collection
          A callable returning a dict-based collection of entities (usually
          an object attribute managed by a SQLAlchemy relation())

        creator
          A function that creates new target entities.  Given two parameters:
          key and value.  The assertion is assumed:
            obj = creator(somekey, somevalue)
            assert getter(somekey) == somevalue

        getter
          A function.  Given an associated object and a key, return the 'value'.

        setter
          A function.  Given an associated object, a key and a value, store
          that value on the object.
        """

        self.lazy_collection = lazy_collection
        self.creator = creator
        self.getter = getter
        self.setter = setter

    col = property(lambda self: self.lazy_collection())

    def _create(self, key, value):
        return self.creator(key, value)

    def _get(self, object):
        return self.getter(object)

    def _set(self, object, key, value):
        return self.setter(object, key, value)

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

    def __nonzero__(self):
        if self.col:
            return True
        else:
            return False

    def __getitem__(self, key):
        return self._get(self.col[key])

    def __setitem__(self, key, value):
        if key in self.col:
            self._set(self.col[key], key, value)
        else:
            self.col[key] = self._create(key, value)

    def __delitem__(self, key):
        del self.col[key]

    def __contains__(self, key):
        # testlib.pragma exempt:__hash__
        return key in self.col
    has_key = __contains__

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

    def clear(self):
        self.col.clear()

    def __eq__(self, other): return dict(self) == other
    def __ne__(self, other): return dict(self) != other
    def __lt__(self, other): return dict(self) < other
    def __le__(self, other): return dict(self) <= other
    def __gt__(self, other): return dict(self) > other
    def __ge__(self, other): return dict(self) >= other
    def __cmp__(self, other): return cmp(dict(self), other)

    def __repr__(self):
        return repr(dict(self.items()))

    def get(self, key, default=None):
        try:
            return self[key]
        except KeyError:
            return default

    def setdefault(self, key, default=None):
        if key not in self.col:
            self.col[key] = self._create(key, default)
            return default
        else:
            return self[key]

    def keys(self):
        return self.col.keys()
    def iterkeys(self):
        return self.col.iterkeys()

    def values(self):
        return [ self._get(member) for member in self.col.values() ]
    def itervalues(self):
        for key in self.col:
            yield self._get(self.col[key])
        raise StopIteration

    def items(self):
        return [(k, self._get(self.col[k])) for k in self]
    def iteritems(self):
        for key in self.col:
            yield (key, self._get(self.col[key]))
        raise StopIteration

    def pop(self, key, default=_NotProvided):
        if default is _NotProvided:
            member = self.col.pop(key)
        else:
            member = self.col.pop(key, default)
        return self._get(member)

    def popitem(self):
        item = self.col.popitem()
        return (item[0], self._get(item[1]))

    def update(self, *a, **kw):
        if len(a) > 1:
            raise TypeError('update expected at most 1 arguments, got %i' %
                            len(a))
        elif len(a) == 1:
            seq_or_map = a[0]
            for item in seq_or_map:
                if isinstance(item, tuple):
                    self[item[0]] = item[1]
                else:
                    self[item] = seq_or_map[item]

        for key, value in kw:
            self[key] = value

    def copy(self):
        return dict(self.items())

    def __hash__(self):
        raise TypeError("%s objects are unhashable" % type(self).__name__)

class _AssociationSet(object):
    """Generic proxying list which proxies set operations to a another set,
    converting association objects to and from a simplified value.
    """

    def __init__(self, lazy_collection, creator, getter, setter):
        """
        collection
          A callable returning a set-based collection of entities (usually an
          object attribute managed by a SQLAlchemy relation())

        creator
          A function that creates new target entities.  Given one parameter:
          value.  The assertion is assumed:
            obj = creator(somevalue)
            assert getter(obj) == somevalue

        getter
          A function.  Given an associated object, return the 'value'.

        setter
          A function.  Given an associated object and a value, store
          that value on the object.
        """

        self.lazy_collection = lazy_collection
        self.creator = creator
        self.getter = getter
        self.setter = setter

    col = property(lambda self: self.lazy_collection())

    def _create(self, value):
        return self.creator(value)

    def _get(self, object):
        return self.getter(object)

    def _set(self, object, value):
        return self.setter(object, value)

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

    def __nonzero__(self):
        if self.col:
            return True
        else:
            return False

    def __contains__(self, value):
        for member in self.col:
            # testlib.pragma exempt:__eq__
            if self._get(member) == value:
                return True
        return False

    def __iter__(self):
        """Iterate over proxied values.  For the actual domain objects,
        iterate over .col instead or just use the underlying collection
        directly from its property on the parent."""
        for member in self.col:
            yield self._get(member)
        raise StopIteration

    def add(self, value):
        if value not in self:
            self.col.add(self._create(value))

    # for discard and remove, choosing a more expensive check strategy rather
    # than call self.creator()
    def discard(self, value):
        for member in self.col:
            if self._get(member) == value:
                self.col.discard(member)
                break

    def remove(self, value):
        for member in self.col:
            if self._get(member) == value:
                self.col.discard(member)
                return
        raise KeyError(value)

    def pop(self):
        if not self.col:
            raise KeyError('pop from an empty set')
        member = self.col.pop()
        return self._get(member)

    def update(self, other):
        for value in other:
            self.add(value)

    def __ior__(self, other):
        if util.duck_type_collection(other) is not util.Set:
            return NotImplemented
        for value in other:
            self.add(value)
        return self

    def _set(self):
        return util.Set(iter(self))

    def union(self, other):
        return util.Set(self).union(other)

    __or__ = union

    def difference(self, other):
        return util.Set(self).difference(other)

    __sub__ = difference

    def difference_update(self, other):
        for value in other:
            self.discard(value)

    def __isub__(self, other):
        if util.duck_type_collection(other) is not util.Set:
            return NotImplemented
        for value in other:
            self.discard(value)
        return self

    def intersection(self, other):
        return util.Set(self).intersection(other)

    __and__ = intersection

    def intersection_update(self, other):
        want, have = self.intersection(other), util.Set(self)

        remove, add = have - want, want - have

        for value in remove:
            self.remove(value)
        for value in add:
            self.add(value)

    def __iand__(self, other):
        if util.duck_type_collection(other) is not util.Set:
            return NotImplemented
        want, have = self.intersection(other), util.Set(self)

        remove, add = have - want, want - have

        for value in remove:
            self.remove(value)
        for value in add:
            self.add(value)
        return self

    def symmetric_difference(self, other):
        return util.Set(self).symmetric_difference(other)

    __xor__ = symmetric_difference

    def symmetric_difference_update(self, other):
        want, have = self.symmetric_difference(other), util.Set(self)

        remove, add = have - want, want - have

        for value in remove:
            self.remove(value)
        for value in add:
            self.add(value)

    def __ixor__(self, other):
        if util.duck_type_collection(other) is not util.Set:
            return NotImplemented
        want, have = self.symmetric_difference(other), util.Set(self)

        remove, add = have - want, want - have

        for value in remove:
            self.remove(value)
        for value in add:
            self.add(value)
        return self

    def issubset(self, other):
        return util.Set(self).issubset(other)

    def issuperset(self, other):
        return util.Set(self).issuperset(other)

    def clear(self):
        self.col.clear()

    def copy(self):
        return util.Set(self)

    def __eq__(self, other): return util.Set(self) == other
    def __ne__(self, other): return util.Set(self) != other
    def __lt__(self, other): return util.Set(self) < other
    def __le__(self, other): return util.Set(self) <= other
    def __gt__(self, other): return util.Set(self) > other
    def __ge__(self, other): return util.Set(self) >= other

    def __repr__(self):
        return repr(util.Set(self))

    def __hash__(self):
        raise TypeError("%s objects are unhashable" % type(self).__name__)