utils.collections

Custom maps, sets, sequences, and other data structures.

Module Contents

Classes

LazyObject()
OrderedDict() Dict where insertion order matters.
AttributeDictMixin() Mixin for Mapping interface that adds attribute access.
AttributeDict() Dict subclass with attribute access.
DictAttribute(self,obj) Dict interface to attributes.
ChainMap(self,*maps,**kwargs) Key lookup on a sequence of maps.
ConfigurationView(self,changes,defaults=None,keys=None,prefix=None) A view over an applications configuration dictionaries.
LimitedSet(self,maxlen=0,expires=0,data=None,minlen=0) Kind-of Set (or priority queue) with limitations.
Evictable() Mixin for classes supporting the evict method.
Messagebuffer(self,maxsize,iterable=None,deque=deque) A buffer of pending messages.
BufferMap(self,maxsize,iterable=None,bufmaxsize=1000) Map of buffers.

Functions

force_mapping(m) Wrap object into supporting the mapping interface if necessary.
lpmerge(L,R) In place left precedent dictionary merge.
class LazyObject
force_mapping(m)

Wrap object into supporting the mapping interface if necessary.

lpmerge(L, R)

In place left precedent dictionary merge.

Keeps values from L, if the value in R is None.

class OrderedDict

Dict where insertion order matters.

_LRUkey()
_LRUkey()
_LRUkey()
move_to_end(key, last=True)
move_to_end(key, last=True)
class AttributeDictMixin

Mixin for Mapping interface that adds attribute access.

I.e., d.key -> d[key]).

__getattr__(k)

d.key -> d[key].

__setattr__(key, value)

d[key] = value -> d.key = value.

class AttributeDict

Dict subclass with attribute access.

class DictAttribute(obj)

Dict interface to attributes.

obj[k] -> obj.k obj[k] = val -> obj.k = val

__init__(obj)
__getattr__(key)
__setattr__(key, value)
get(key, default=None)
setdefault(key, default=None)
__getitem__(key)
__setitem__(key, value)
__contains__(key)
_iterate_keys()
__iter__()
_iterate_items()
_iterate_values()
keys()
items()
values()
class ChainMap(*maps, **kwargs)

Key lookup on a sequence of maps.

__init__(*maps, **kwargs)
add_defaults(d)
pop(key, *default)
__missing__(key)
_key(key)
__getitem__(key)
__setitem__(key, value)
__delitem__(key)
clear()
get(key, default=None)
__len__()
__iter__()
__contains__(key)
__bool__()
setdefault(key, default=None)
update(*args, **kwargs)
__repr__()
fromkeys(iterable, *args)

Create a ChainMap with a single dict created from the iterable.

copy()
_iter(op)
_iterate_keys()
_iterate_items()
_iterate_values()
keys()
items()
values()
class ConfigurationView(changes, defaults=None, keys=None, prefix=None)

A view over an applications configuration dictionaries.

Custom (but older) version of collections.ChainMap.

If the key does not exist in changes, the defaults dictionaries are consulted.

Arguments:

changes (Mapping): Map of configuration changes. defaults (List[Mapping]): List of dictionaries containing

the default configuration.
__init__(changes, defaults=None, keys=None, prefix=None)
_to_keys(key)
__getitem__(key)
__setitem__(key, value)
first(*keys)
get(key, default=None)
clear()

Remove all changes, but keep defaults.

__contains__(key)
swap_with(other)
class LimitedSet(maxlen=0, expires=0, data=None, minlen=0)

Kind-of Set (or priority queue) with limitations.

Good for when you need to test for membership (a in set), but the set should not grow unbounded.

maxlen is enforced at all times, so if the limit is reached we’ll also remove non-expired items.

You can also configure minlen: this is the minimal residual size of the set.

All arguments are optional, and no limits are enabled by default.

Arguments:
maxlen (int): Optional max number of items.
Adding more items than maxlen will result in immediate removal of items sorted by oldest insertion time.
expires (float): TTL for all items.
Expired items are purged as keys are inserted.
minlen (int): Minimal residual size of this set.

New in version 4.0.

Value must be less than maxlen if both are configured.

Older expired items will be deleted, only after the set exceeds minlen number of items.

data (Sequence): Initial data to initialize set with.
Can be an iterable of (key, value) pairs, a dict ({key: insertion_time}), or another instance of LimitedSet.
Example:
>>> s = LimitedSet(maxlen=50000, expires=3600, minlen=4000)
>>> for i in range(60000):
...     s.add(i)
...     s.add(str(i))
...
>>> 57000 in s  # last 50k inserted values are kept
True
>>> '10' in s  # '10' did expire and was purged from set.
False
>>> len(s)  # maxlen is reached
50000
>>> s.purge(now=monotonic() + 7200)  # clock + 2 hours
>>> len(s)  # now only minlen items are cached
4000
>>>> 57000 in s  # even this item is gone now
False
__init__(maxlen=0, expires=0, data=None, minlen=0)
_refresh_heap()

Time consuming recreating of heap. Don’t run this too often.

_maybe_refresh_heap()
clear()

Clear all data, start from scratch again.

add(item, now=None)

Add a new item, or reset the expiry time of an existing item.

update(other)

Update this set from other LimitedSet, dict or iterable.

discard(item)
purge(now=None)

Check oldest items and remove them if needed.

Arguments:
now (float): Time of purging – by default right now.
This can be useful for unit testing.
pop(default=None)

Remove and return the oldest item, or None when empty.

as_dict()

Whole set as serializable dictionary.

Example:
>>> s = LimitedSet(maxlen=200)
>>> r = LimitedSet(maxlen=200)
>>> for i in range(500):
...     s.add(i)
...
>>> r.update(s.as_dict())
>>> r == s
True
__eq__(other)
__ne__(other)
__repr__()
__iter__()
__len__()
__contains__(key)
__reduce__()
__bool__()
_heap_overload()

Compute how much is heap bigger than data [percents].

class Evictable

Mixin for classes supporting the evict method.

evict()

Force evict until maxsize is enforced.

_evict(limit=100, range=range)
_evict1()
class Messagebuffer(maxsize, iterable=None, deque=deque)

A buffer of pending messages.

__init__(maxsize, iterable=None, deque=deque)
put(item)
extend(it)
take(*default)
_pop_to_evict()
__repr__()
__iter__()
__len__()
__contains__(item)
__reversed__()
__getitem__(index)
_evictcount()
class BufferMap(maxsize, iterable=None, bufmaxsize=1000)

Map of buffers.

__init__(maxsize, iterable=None, bufmaxsize=1000)
put(key, item)
extend(key, it)
take(key, *default)
_get_or_create_buffer(key)
_new_buffer()
_LRUpop(*default)
_pop_to_evict()
__repr__()
_evictcount()