def _asdict(self):
return dict((key, self.__dict__[key]) for key in self.keys())
+
class ImmutableContainer(object):
def _immutable(self, *arg, **kw):
raise TypeError("%s object is immutable" % self.__class__.__name__)
__delitem__ = __setitem__ = __setattr__ = _immutable
+
class immutabledict(ImmutableContainer, dict):
clear = pop = popitem = setdefault = \
def __repr__(self):
return "immutabledict(%s)" % dict.__repr__(self)
+
class Properties(object):
"""Provide a __getattr__/__setattr__ interface over a dict."""
def clear(self):
self._data.clear()
+
class OrderedProperties(Properties):
"""Provide a __getattr__/__setattr__ interface with an OrderedDict
as backing store."""
self._list.remove(item[0])
return item
+
class OrderedSet(set):
def __init__(self, d=None):
set.__init__(self)
def intersection_update(self, other):
other = set(other)
set.intersection_update(self, other)
- self._list = [ a for a in self._list if a in other]
+ self._list = [a for a in self._list if a in other]
return self
__iand__ = intersection_update
def symmetric_difference_update(self, other):
set.symmetric_difference_update(self, other)
- self._list = [ a for a in self._list if a in self]
- self._list += [ a for a in other._list if a in self]
+ self._list = [a for a in self._list if a in self]
+ self._list += [a for a in other._list if a in self]
return self
__ixor__ = symmetric_difference_update
def difference_update(self, other):
set.difference_update(self, other)
- self._list = [ a for a in self._list if a in self]
+ self._list = [a for a in self._list if a in self]
return self
__isub__ = difference_update
def union(self, iterable):
result = type(self)()
# testlib.pragma exempt:__hash__
- result._members.update(
- self._working_set(self._member_id_tuples()).union(_iter_id(iterable)))
+ members = self._member_id_tuples()
+ other = _iter_id(iterable)
+ result._members.update(self._working_set(members).union(other))
return result
def __or__(self, other):
def difference(self, iterable):
result = type(self)()
# testlib.pragma exempt:__hash__
- result._members.update(
- self._working_set(self._member_id_tuples()).difference(_iter_id(iterable)))
+ members = self._member_id_tuples()
+ other = _iter_id(iterable)
+ result._members.update(self._working_set(members).difference(other))
return result
def __sub__(self, other):
def intersection(self, iterable):
result = type(self)()
# testlib.pragma exempt:__hash__
- result._members.update(
- self._working_set(self._member_id_tuples()).intersection(_iter_id(iterable)))
+ members = self._member_id_tuples()
+ other = _iter_id(iterable)
+ result._members.update(self._working_set(members).intersection(other))
return result
def __and__(self, other):
def symmetric_difference(self, iterable):
result = type(self)()
# testlib.pragma exempt:__hash__
+ members = self._member_id_tuples()
+ other = _iter_id(iterable)
result._members.update(
- self._working_set(self._member_id_tuples()).symmetric_difference(_iter_id(iterable)))
+ self._working_set(members).symmetric_difference(other))
return result
def _member_id_tuples(self):
def __repr__(self):
return '%s(%r)' % (type(self).__name__, self._members.values())
+
class WeakSequence(object):
def __init__(self, elements):
self._storage = weakref.WeakValueDictionary(
except KeyError:
raise IndexError("Index %s out of range" % index)
+
class OrderedIdentitySet(IdentitySet):
class _working_set(OrderedSet):
# a testing pragma: exempt the OIDS working set from the test suite's
ordered_column_set = OrderedSet
populate_column_dict = PopulateDict
+
def unique_list(seq, hashfunc=None):
seen = {}
if not hashfunc:
if hashfunc(x) not in seen
and not seen.__setitem__(hashfunc(x), True)]
+
class UniqueAppender(object):
"""Appends items to a collection ensuring uniqueness.
def __iter__(self):
return iter(self.data)
+
def to_list(x, default=None):
if x is None:
return default
else:
return x
+
def to_set(x):
if x is None:
return set()
else:
return x
+
def to_column_set(x):
if x is None:
return column_set()
else:
return x
+
def update_copy(d, _new=None, **kw):
"""Copy the given dict and update with the given values."""
d.update(**kw)
return d
+
def flatten_iterator(x):
"""Given an iterator of which further sub-elements may also be
iterators, flatten the sub-elements into a single iterator.
yield elem
-
class LRUCache(dict):
"""Dictionary with 'squishy' removal of least
recently used items.
except KeyError:
pass
+
class ThreadLocalRegistry(ScopedRegistry):
"""A :class:`.ScopedRegistry` that uses a ``threading.local()``
variable for storage.
except AttributeError:
pass
+
def _iter_id(iterable):
"""Generator: ((id(o), o) for o in iterable)."""
for item in iterable:
yield id(item), item
-