Source code
Revision control
Copy as Markdown
Other Tools
import sys
from array import array
from collections import abc
from ._abc import MultiMapping, MutableMultiMapping
_marker = object()
class istr(str):
"""Case insensitive str."""
__is_istr__ = True
upstr = istr # for relaxing backward compatibility problems
def getversion(md):
if not isinstance(md, _Base):
raise TypeError("Parameter should be multidict or proxy")
return md._impl._version
_version = array("Q", [0])
class _Impl:
__slots__ = ("_items", "_version")
def __init__(self):
self._items = []
self.incr_version()
def incr_version(self):
global _version
v = _version
v[0] += 1
self._version = v[0]
if sys.implementation.name != "pypy":
def __sizeof__(self):
return object.__sizeof__(self) + sys.getsizeof(self._items)
class _Base:
def _title(self, key):
return key
def getall(self, key, default=_marker):
"""Return a list of all values matching the key."""
identity = self._title(key)
res = [v for i, k, v in self._impl._items if i == identity]
if res:
return res
if not res and default is not _marker:
return default
raise KeyError("Key not found: %r" % key)
def getone(self, key, default=_marker):
"""Get first value matching the key."""
identity = self._title(key)
for i, k, v in self._impl._items:
if i == identity:
return v
if default is not _marker:
return default
raise KeyError("Key not found: %r" % key)
# Mapping interface #
def __getitem__(self, key):
return self.getone(key)
def get(self, key, default=None):
"""Get first value matching the key.
The method is alias for .getone().
"""
return self.getone(key, default)
def __iter__(self):
return iter(self.keys())
def __len__(self):
return len(self._impl._items)
def keys(self):
"""Return a new view of the dictionary's keys."""
return _KeysView(self._impl)
def items(self):
"""Return a new view of the dictionary's items *(key, value) pairs)."""
return _ItemsView(self._impl)
def values(self):
"""Return a new view of the dictionary's values."""
return _ValuesView(self._impl)
def __eq__(self, other):
if not isinstance(other, abc.Mapping):
return NotImplemented
if isinstance(other, _Base):
lft = self._impl._items
rht = other._impl._items
if len(lft) != len(rht):
return False
for (i1, k2, v1), (i2, k2, v2) in zip(lft, rht):
if i1 != i2 or v1 != v2:
return False
return True
if len(self._impl._items) != len(other):
return False
for k, v in self.items():
nv = other.get(k, _marker)
if v != nv:
return False
return True
def __contains__(self, key):
identity = self._title(key)
for i, k, v in self._impl._items:
if i == identity:
return True
return False
def __repr__(self):
body = ", ".join("'{}': {!r}".format(k, v) for k, v in self.items())
return "<{}({})>".format(self.__class__.__name__, body)
class MultiDictProxy(_Base, MultiMapping):
"""Read-only proxy for MultiDict instance."""
def __init__(self, arg):
if not isinstance(arg, (MultiDict, MultiDictProxy)):
raise TypeError(
"ctor requires MultiDict or MultiDictProxy instance"
", not {}".format(type(arg))
)
self._impl = arg._impl
def __reduce__(self):
raise TypeError("can't pickle {} objects".format(self.__class__.__name__))
def copy(self):
"""Return a copy of itself."""
return MultiDict(self.items())
class CIMultiDictProxy(MultiDictProxy):
"""Read-only proxy for CIMultiDict instance."""
def __init__(self, arg):
if not isinstance(arg, (CIMultiDict, CIMultiDictProxy)):
raise TypeError(
"ctor requires CIMultiDict or CIMultiDictProxy instance"
", not {}".format(type(arg))
)
self._impl = arg._impl
def _title(self, key):
return key.title()
def copy(self):
"""Return a copy of itself."""
return CIMultiDict(self.items())
class MultiDict(_Base, MutableMultiMapping):
"""Dictionary with the support for duplicate keys."""
def __init__(self, *args, **kwargs):
self._impl = _Impl()
self._extend(args, kwargs, self.__class__.__name__, self._extend_items)
if sys.implementation.name != "pypy":
def __sizeof__(self):
return object.__sizeof__(self) + sys.getsizeof(self._impl)
def __reduce__(self):
return (self.__class__, (list(self.items()),))
def _title(self, key):
return key
def _key(self, key):
if isinstance(key, str):
return key
else:
raise TypeError(
"MultiDict keys should be either str " "or subclasses of str"
)
def add(self, key, value):
identity = self._title(key)
self._impl._items.append((identity, self._key(key), value))
self._impl.incr_version()
def copy(self):
"""Return a copy of itself."""
cls = self.__class__
return cls(self.items())
__copy__ = copy
def extend(self, *args, **kwargs):
"""Extend current MultiDict with more values.
This method must be used instead of update.
"""
self._extend(args, kwargs, "extend", self._extend_items)
def _extend(self, args, kwargs, name, method):
if len(args) > 1:
raise TypeError(
"{} takes at most 1 positional argument"
" ({} given)".format(name, len(args))
)
if args:
arg = args[0]
if isinstance(args[0], (MultiDict, MultiDictProxy)) and not kwargs:
items = arg._impl._items
else:
if hasattr(arg, "items"):
arg = arg.items()
if kwargs:
arg = list(arg)
arg.extend(list(kwargs.items()))
items = []
for item in arg:
if not len(item) == 2:
raise TypeError(
"{} takes either dict or list of (key, value) "
"tuples".format(name)
)
items.append((self._title(item[0]), self._key(item[0]), item[1]))
method(items)
else:
method(
[
(self._title(key), self._key(key), value)
for key, value in kwargs.items()
]
)
def _extend_items(self, items):
for identity, key, value in items:
self.add(key, value)
def clear(self):
"""Remove all items from MultiDict."""
self._impl._items.clear()
self._impl.incr_version()
# Mapping interface #
def __setitem__(self, key, value):
self._replace(key, value)
def __delitem__(self, key):
identity = self._title(key)
items = self._impl._items
found = False
for i in range(len(items) - 1, -1, -1):
if items[i][0] == identity:
del items[i]
found = True
if not found:
raise KeyError(key)
else:
self._impl.incr_version()
def setdefault(self, key, default=None):
"""Return value for key, set value to default if key is not present."""
identity = self._title(key)
for i, k, v in self._impl._items:
if i == identity:
return v
self.add(key, default)
return default
def popone(self, key, default=_marker):
"""Remove specified key and return the corresponding value.
If key is not found, d is returned if given, otherwise
KeyError is raised.
"""
identity = self._title(key)
for i in range(len(self._impl._items)):
if self._impl._items[i][0] == identity:
value = self._impl._items[i][2]
del self._impl._items[i]
self._impl.incr_version()
return value
if default is _marker:
raise KeyError(key)
else:
return default
pop = popone # type: ignore
def popall(self, key, default=_marker):
"""Remove all occurrences of key and return the list of corresponding
values.
If key is not found, default is returned if given, otherwise
KeyError is raised.
"""
found = False
identity = self._title(key)
ret = []
for i in range(len(self._impl._items) - 1, -1, -1):
item = self._impl._items[i]
if item[0] == identity:
ret.append(item[2])
del self._impl._items[i]
self._impl.incr_version()
found = True
if not found:
if default is _marker:
raise KeyError(key)
else:
return default
else:
ret.reverse()
return ret
def popitem(self):
"""Remove and return an arbitrary (key, value) pair."""
if self._impl._items:
i = self._impl._items.pop(0)
self._impl.incr_version()
return i[1], i[2]
else:
raise KeyError("empty multidict")
def update(self, *args, **kwargs):
"""Update the dictionary from *other*, overwriting existing keys."""
self._extend(args, kwargs, "update", self._update_items)
def _update_items(self, items):
if not items:
return
used_keys = {}
for identity, key, value in items:
start = used_keys.get(identity, 0)
for i in range(start, len(self._impl._items)):
item = self._impl._items[i]
if item[0] == identity:
used_keys[identity] = i + 1
self._impl._items[i] = (identity, key, value)
break
else:
self._impl._items.append((identity, key, value))
used_keys[identity] = len(self._impl._items)
# drop tails
i = 0
while i < len(self._impl._items):
item = self._impl._items[i]
identity = item[0]
pos = used_keys.get(identity)
if pos is None:
i += 1
continue
if i >= pos:
del self._impl._items[i]
else:
i += 1
self._impl.incr_version()
def _replace(self, key, value):
key = self._key(key)
identity = self._title(key)
items = self._impl._items
for i in range(len(items)):
item = items[i]
if item[0] == identity:
items[i] = (identity, key, value)
# i points to last found item
rgt = i
self._impl.incr_version()
break
else:
self._impl._items.append((identity, key, value))
self._impl.incr_version()
return
# remove all tail items
i = rgt + 1
while i < len(items):
item = items[i]
if item[0] == identity:
del items[i]
else:
i += 1
class CIMultiDict(MultiDict):
"""Dictionary with the support for duplicate case-insensitive keys."""
def _title(self, key):
return key.title()
class _Iter:
__slots__ = ("_size", "_iter")
def __init__(self, size, iterator):
self._size = size
self._iter = iterator
def __iter__(self):
return self
def __next__(self):
return next(self._iter)
def __length_hint__(self):
return self._size
class _ViewBase:
def __init__(self, impl):
self._impl = impl
self._version = impl._version
def __len__(self):
return len(self._impl._items)
class _ItemsView(_ViewBase, abc.ItemsView):
def __contains__(self, item):
assert isinstance(item, tuple) or isinstance(item, list)
assert len(item) == 2
for i, k, v in self._impl._items:
if item[0] == k and item[1] == v:
return True
return False
def __iter__(self):
return _Iter(len(self), self._iter())
def _iter(self):
for i, k, v in self._impl._items:
if self._version != self._impl._version:
raise RuntimeError("Dictionary changed during iteration")
yield k, v
def __repr__(self):
lst = []
for item in self._impl._items:
lst.append("{!r}: {!r}".format(item[1], item[2]))
body = ", ".join(lst)
return "{}({})".format(self.__class__.__name__, body)
class _ValuesView(_ViewBase, abc.ValuesView):
def __contains__(self, value):
for item in self._impl._items:
if item[2] == value:
return True
return False
def __iter__(self):
return _Iter(len(self), self._iter())
def _iter(self):
for item in self._impl._items:
if self._version != self._impl._version:
raise RuntimeError("Dictionary changed during iteration")
yield item[2]
def __repr__(self):
lst = []
for item in self._impl._items:
lst.append("{!r}".format(item[2]))
body = ", ".join(lst)
return "{}({})".format(self.__class__.__name__, body)
class _KeysView(_ViewBase, abc.KeysView):
def __contains__(self, key):
for item in self._impl._items:
if item[1] == key:
return True
return False
def __iter__(self):
return _Iter(len(self), self._iter())
def _iter(self):
for item in self._impl._items:
if self._version != self._impl._version:
raise RuntimeError("Dictionary changed during iteration")
yield item[1]
def __repr__(self):
lst = []
for item in self._impl._items:
lst.append("{!r}".format(item[1]))
body = ", ".join(lst)
return "{}({})".format(self.__class__.__name__, body)