_collections_abc.py

# Copyright 2007 Google, Inc. All Rights Reserved.
# Licensed to PSF under a Contributor Agreement.

"""Abstract Base Classes (ABCs) for collections, according to PEP 3119.

Unit tests are in test_collections.
"""

############ Maintenance notes #########################################
#
# ABCs are different from other standard library modules in that they
# specify compliance tests. In general, once an ABC has been published,
# new methods (either abstract or concrete) cannot be added.
#
# Though classes that inherit from an ABC would automatically receive a
# new mixin method, registered classes would become non-compliant and
# violate the contract promised by ``isinstance(someobj, SomeABC)``.
#
# Though irritating, the correct procedure for adding new abstract or
# mixin methods is to create a new ABC as a subclass of the previous
# ABC. For example, union(), intersection(), and difference() cannot
# be added to Set but could go into a new ABC that extends Set.
#
# Because they are so hard to change, new ABCs should have their APIs
# carefully thought through prior to publication.
#
# Since ABCMeta only checks for the presence of methods, it is possible
# to alter the signature of a method by adding optional arguments
# or changing parameters names. This is still a bit dubious but at
# least it won't cause isinstance() to return an incorrect result.
#
#
#######################################################################

fromabcimportABCMeta, abstractmethod
importsys

GenericAlias=type(list[int])
EllipsisType=type(...)
def_f(): pass
FunctionType=type(_f)
del_f

__all__= ["Awaitable", "Coroutine",
"AsyncIterable", "AsyncIterator", "AsyncGenerator",
"Hashable", "Iterable", "Iterator", "Generator", "Reversible",
"Sized", "Container", "Callable", "Collection",
"Set", "MutableSet",
"Mapping", "MutableMapping",
"MappingView", "KeysView", "ItemsView", "ValuesView",
"Sequence", "MutableSequence",
"ByteString", "Buffer",
 ]

# This module has been renamed from collections.abc to _collections_abc to
# speed up interpreter startup. Some of the types such as MutableMapping are
# required early but collections module imports a lot of other modules.
# See issue #19218
__name__="collections.abc"

# Private list of types that we want to register with the various ABCs
# so that they will pass tests like:
# it = iter(somebytearray)
# assert isinstance(it, Iterable)
# Note: in other implementations, these types might not be distinct
# and they may have their own implementation specific types that
# are not included on this list.
bytes_iterator=type(iter(b''))
bytearray_iterator=type(iter(bytearray()))
#callable_iterator = ???
dict_keyiterator=type(iter({}.keys()))
dict_valueiterator=type(iter({}.values()))
dict_itemiterator=type(iter({}.items()))
list_iterator=type(iter([]))
list_reverseiterator=type(iter(reversed([])))
range_iterator=type(iter(range(0)))
longrange_iterator=type(iter(range(1<<1000)))
set_iterator=type(iter(set()))
str_iterator=type(iter(""))
tuple_iterator=type(iter(()))
zip_iterator=type(iter(zip()))
## views ##
dict_keys=type({}.keys())
dict_values=type({}.values())
dict_items=type({}.items())
## misc ##
mappingproxy=type(type.__dict__)
def_get_framelocalsproxy():
returntype(sys._getframe().f_locals)
framelocalsproxy=_get_framelocalsproxy()
del_get_framelocalsproxy
generator=type((lambda: (yield))())
## coroutine ##
asyncdef_coro(): pass
_coro=_coro()
coroutine=type(_coro)
_coro.close() # Prevent ResourceWarning
del_coro
## asynchronous generator ##
asyncdef_ag(): yield
_ag=_ag()
async_generator=type(_ag)
del_ag


### ONE-TRICK PONIES ###

def_check_methods(C, *methods):
mro=C.__mro__
formethodinmethods:
forBinmro:
ifmethodinB.__dict__:
ifB.__dict__[method] isNone:
returnNotImplemented
break
else:
returnNotImplemented
returnTrue

classHashable(metaclass=ABCMeta):

__slots__= ()

@abstractmethod
def__hash__(self):
return0

@classmethod
def__subclasshook__(cls, C):
ifclsisHashable:
return_check_methods(C, "__hash__")
returnNotImplemented


classAwaitable(metaclass=ABCMeta):

__slots__= ()

@abstractmethod
def__await__(self):
yield

@classmethod
def__subclasshook__(cls, C):
ifclsisAwaitable:
return_check_methods(C, "__await__")
returnNotImplemented

__class_getitem__=classmethod(GenericAlias)


classCoroutine(Awaitable):

__slots__= ()

@abstractmethod
defsend(self, value):
"""Send a value into the coroutine.
 Return next yielded value or raise StopIteration.
 """
raiseStopIteration

@abstractmethod
defthrow(self, typ, val=None, tb=None):
"""Raise an exception in the coroutine.
 Return next yielded value or raise StopIteration.
 """
ifvalisNone:
iftbisNone:
raisetyp
val=typ()
iftbisnotNone:
val=val.with_traceback(tb)
raiseval

defclose(self):
"""Raise GeneratorExit inside coroutine.
 """
try:
self.throw(GeneratorExit)
except (GeneratorExit, StopIteration):
pass
else:
raiseRuntimeError("coroutine ignored GeneratorExit")

@classmethod
def__subclasshook__(cls, C):
ifclsisCoroutine:
return_check_methods(C, '__await__', 'send', 'throw', 'close')
returnNotImplemented


Coroutine.register(coroutine)


classAsyncIterable(metaclass=ABCMeta):

__slots__= ()

@abstractmethod
def__aiter__(self):
returnAsyncIterator()

@classmethod
def__subclasshook__(cls, C):
ifclsisAsyncIterable:
return_check_methods(C, "__aiter__")
returnNotImplemented

__class_getitem__=classmethod(GenericAlias)


classAsyncIterator(AsyncIterable):

__slots__= ()

@abstractmethod
asyncdef__anext__(self):
"""Return the next item or raise StopAsyncIteration when exhausted."""
raiseStopAsyncIteration

def__aiter__(self):
returnself

@classmethod
def__subclasshook__(cls, C):
ifclsisAsyncIterator:
return_check_methods(C, "__anext__", "__aiter__")
returnNotImplemented


classAsyncGenerator(AsyncIterator):

__slots__= ()

asyncdef__anext__(self):
"""Return the next item from the asynchronous generator.
 When exhausted, raise StopAsyncIteration.
 """
returnawaitself.asend(None)

@abstractmethod
asyncdefasend(self, value):
"""Send a value into the asynchronous generator.
 Return next yielded value or raise StopAsyncIteration.
 """
raiseStopAsyncIteration

@abstractmethod
asyncdefathrow(self, typ, val=None, tb=None):
"""Raise an exception in the asynchronous generator.
 Return next yielded value or raise StopAsyncIteration.
 """
ifvalisNone:
iftbisNone:
raisetyp
val=typ()
iftbisnotNone:
val=val.with_traceback(tb)
raiseval

asyncdefaclose(self):
"""Raise GeneratorExit inside coroutine.
 """
try:
awaitself.athrow(GeneratorExit)
except (GeneratorExit, StopAsyncIteration):
pass
else:
raiseRuntimeError("asynchronous generator ignored GeneratorExit")

@classmethod
def__subclasshook__(cls, C):
ifclsisAsyncGenerator:
return_check_methods(C, '__aiter__', '__anext__',
'asend', 'athrow', 'aclose')
returnNotImplemented


AsyncGenerator.register(async_generator)


classIterable(metaclass=ABCMeta):

__slots__= ()

@abstractmethod
def__iter__(self):
whileFalse:
yieldNone

@classmethod
def__subclasshook__(cls, C):
ifclsisIterable:
return_check_methods(C, "__iter__")
returnNotImplemented

__class_getitem__=classmethod(GenericAlias)


classIterator(Iterable):

__slots__= ()

@abstractmethod
def__next__(self):
'Return the next item from the iterator. When exhausted, raise StopIteration'
raiseStopIteration

def__iter__(self):
returnself

@classmethod
def__subclasshook__(cls, C):
ifclsisIterator:
return_check_methods(C, '__iter__', '__next__')
returnNotImplemented


Iterator.register(bytes_iterator)
Iterator.register(bytearray_iterator)
#Iterator.register(callable_iterator)
Iterator.register(dict_keyiterator)
Iterator.register(dict_valueiterator)
Iterator.register(dict_itemiterator)
Iterator.register(list_iterator)
Iterator.register(list_reverseiterator)
Iterator.register(range_iterator)
Iterator.register(longrange_iterator)
Iterator.register(set_iterator)
Iterator.register(str_iterator)
Iterator.register(tuple_iterator)
Iterator.register(zip_iterator)


classReversible(Iterable):

__slots__= ()

@abstractmethod
def__reversed__(self):
whileFalse:
yieldNone

@classmethod
def__subclasshook__(cls, C):
ifclsisReversible:
return_check_methods(C, "__reversed__", "__iter__")
returnNotImplemented


classGenerator(Iterator):

__slots__= ()

def__next__(self):
"""Return the next item from the generator.
 When exhausted, raise StopIteration.
 """
returnself.send(None)

@abstractmethod
defsend(self, value):
"""Send a value into the generator.
 Return next yielded value or raise StopIteration.
 """
raiseStopIteration

@abstractmethod
defthrow(self, typ, val=None, tb=None):
"""Raise an exception in the generator.
 Return next yielded value or raise StopIteration.
 """
ifvalisNone:
iftbisNone:
raisetyp
val=typ()
iftbisnotNone:
val=val.with_traceback(tb)
raiseval

defclose(self):
"""Raise GeneratorExit inside generator.
 """
try:
self.throw(GeneratorExit)
except (GeneratorExit, StopIteration):
pass
else:
raiseRuntimeError("generator ignored GeneratorExit")

@classmethod
def__subclasshook__(cls, C):
ifclsisGenerator:
return_check_methods(C, '__iter__', '__next__',
'send', 'throw', 'close')
returnNotImplemented


Generator.register(generator)


classSized(metaclass=ABCMeta):

__slots__= ()

@abstractmethod
def__len__(self):
return0

@classmethod
def__subclasshook__(cls, C):
ifclsisSized:
return_check_methods(C, "__len__")
returnNotImplemented


classContainer(metaclass=ABCMeta):

__slots__= ()

@abstractmethod
def__contains__(self, x):
returnFalse

@classmethod
def__subclasshook__(cls, C):
ifclsisContainer:
return_check_methods(C, "__contains__")
returnNotImplemented

__class_getitem__=classmethod(GenericAlias)


classCollection(Sized, Iterable, Container):

__slots__= ()

@classmethod
def__subclasshook__(cls, C):
ifclsisCollection:
return_check_methods(C, "__len__", "__iter__", "__contains__")
returnNotImplemented


classBuffer(metaclass=ABCMeta):

__slots__= ()

@abstractmethod
def__buffer__(self, flags: int, /) ->memoryview:
raiseNotImplementedError

@classmethod
def__subclasshook__(cls, C):
ifclsisBuffer:
return_check_methods(C, "__buffer__")
returnNotImplemented


class_CallableGenericAlias(GenericAlias):
""" Represent `Callable[argtypes, resulttype]`.

 This sets ``__args__`` to a tuple containing the flattened ``argtypes``
 followed by ``resulttype``.

 Example: ``Callable[[int, str], float]`` sets ``__args__`` to
 ``(int, str, float)``.
 """

__slots__= ()

def__new__(cls, origin, args):
ifnot (isinstance(args, tuple) andlen(args) ==2):
raiseTypeError(
"Callable must be used as Callable[[arg, ...], result].")
t_args, t_result=args
ifisinstance(t_args, (tuple, list)):
args= (*t_args, t_result)
elifnot_is_param_expr(t_args):
raiseTypeError(f"Expected a list of types, an ellipsis, "
f"ParamSpec, or Concatenate. Got {t_args}")
returnsuper().__new__(cls, origin, args)

def__repr__(self):
iflen(self.__args__) ==2and_is_param_expr(self.__args__[0]):
returnsuper().__repr__()
return (f'collections.abc.Callable'
f'[[{", ".join([_type_repr(a) forainself.__args__[:-1]])}], '
f'{_type_repr(self.__args__[-1])}]')

def__reduce__(self):
args=self.__args__
ifnot (len(args) ==2and_is_param_expr(args[0])):
args=list(args[:-1]), args[-1]
return_CallableGenericAlias, (Callable, args)

def__getitem__(self, item):
# Called during TypeVar substitution, returns the custom subclass
# rather than the default types.GenericAlias object. Most of the
# code is copied from typing's _GenericAlias and the builtin
# types.GenericAlias.
ifnotisinstance(item, tuple):
item= (item,)

new_args=super().__getitem__(item).__args__

# args[0] occurs due to things like Z[[int, str, bool]] from PEP 612
ifnotisinstance(new_args[0], (tuple, list)):
t_result=new_args[-1]
t_args=new_args[:-1]
new_args= (t_args, t_result)
return_CallableGenericAlias(Callable, tuple(new_args))

def_is_param_expr(obj):
"""Checks if obj matches either a list of types, ``...``, ``ParamSpec`` or
 ``_ConcatenateGenericAlias`` from typing.py
 """
ifobjisEllipsis:
returnTrue
ifisinstance(obj, list):
returnTrue
obj=type(obj)
names= ('ParamSpec', '_ConcatenateGenericAlias')
returnobj.__module__=='typing'andany(obj.__name__==namefornameinnames)

def_type_repr(obj):
"""Return the repr() of an object, special-casing types (internal helper).

 Copied from :mod:`typing` since collections.abc
 shouldn't depend on that module.
 (Keep this roughly in sync with the typing version.)
 """
ifisinstance(obj, type):
ifobj.__module__=='builtins':
returnobj.__qualname__
returnf'{obj.__module__}.{obj.__qualname__}'
ifobjisEllipsis:
return'...'
ifisinstance(obj, FunctionType):
returnobj.__name__
returnrepr(obj)


classCallable(metaclass=ABCMeta):

__slots__= ()

@abstractmethod
def__call__(self, *args, **kwds):
returnFalse

@classmethod
def__subclasshook__(cls, C):
ifclsisCallable:
return_check_methods(C, "__call__")
returnNotImplemented

__class_getitem__=classmethod(_CallableGenericAlias)


### SETS ###


classSet(Collection):
"""A set is a finite, iterable container.

 This class provides concrete generic implementations of all
 methods except for __contains__, __iter__ and __len__.

 To override the comparisons (presumably for speed, as the
 semantics are fixed), redefine __le__ and __ge__,
 then the other operations will automatically follow suit.
 """

__slots__= ()

def__le__(self, other):
ifnotisinstance(other, Set):
returnNotImplemented
iflen(self) >len(other):
returnFalse
foreleminself:
ifelemnotinother:
returnFalse
returnTrue

def__lt__(self, other):
ifnotisinstance(other, Set):
returnNotImplemented
returnlen(self) <len(other) andself.__le__(other)

def__gt__(self, other):
ifnotisinstance(other, Set):
returnNotImplemented
returnlen(self) >len(other) andself.__ge__(other)

def__ge__(self, other):
ifnotisinstance(other, Set):
returnNotImplemented
iflen(self) <len(other):
returnFalse
foreleminother:
ifelemnotinself:
returnFalse
returnTrue

def__eq__(self, other):
ifnotisinstance(other, Set):
returnNotImplemented
returnlen(self) ==len(other) andself.__le__(other)

@classmethod
def_from_iterable(cls, it):
'''Construct an instance of the class from any iterable input.

 Must override this method if the class constructor signature
 does not accept an iterable for an input.
 '''
returncls(it)

def__and__(self, other):
ifnotisinstance(other, Iterable):
returnNotImplemented
returnself._from_iterable(valueforvalueinotherifvalueinself)

__rand__=__and__

defisdisjoint(self, other):
'Return True if two sets have a null intersection.'
forvalueinother:
ifvalueinself:
returnFalse
returnTrue

def__or__(self, other):
ifnotisinstance(other, Iterable):
returnNotImplemented
chain= (eforsin (self, other) foreins)
returnself._from_iterable(chain)

__ror__=__or__

def__sub__(self, other):
ifnotisinstance(other, Set):
ifnotisinstance(other, Iterable):
returnNotImplemented
other=self._from_iterable(other)
returnself._from_iterable(valueforvalueinself
ifvaluenotinother)

def__rsub__(self, other):
ifnotisinstance(other, Set):
ifnotisinstance(other, Iterable):
returnNotImplemented
other=self._from_iterable(other)
returnself._from_iterable(valueforvalueinother
ifvaluenotinself)

def__xor__(self, other):
ifnotisinstance(other, Set):
ifnotisinstance(other, Iterable):
returnNotImplemented
other=self._from_iterable(other)
return (self-other) | (other-self)

__rxor__=__xor__

def_hash(self):
"""Compute the hash value of a set.

 Note that we don't define __hash__: not all sets are hashable.
 But if you define a hashable set type, its __hash__ should
 call this function.

 This must be compatible __eq__.

 All sets ought to compare equal if they contain the same
 elements, regardless of how they are implemented, and
 regardless of the order of the elements; so there's not much
 freedom for __eq__ or __hash__. We match the algorithm used
 by the built-in frozenset type.
 """
MAX=sys.maxsize
MASK=2*MAX+1
n=len(self)
h=1927868237* (n+1)
h&=MASK
forxinself:
hx=hash(x)
h^= (hx^ (hx<<16) ^89869747) *3644798167
h&=MASK
h^= (h>>11) ^ (h>>25)
h=h*69069+907133923
h&=MASK
ifh>MAX:
h-=MASK+1
ifh==-1:
h=590923713
returnh


Set.register(frozenset)


classMutableSet(Set):
"""A mutable set is a finite, iterable container.

 This class provides concrete generic implementations of all
 methods except for __contains__, __iter__, __len__,
 add(), and discard().

 To override the comparisons (presumably for speed, as the
 semantics are fixed), all you have to do is redefine __le__ and
 then the other operations will automatically follow suit.
 """

__slots__= ()

@abstractmethod
defadd(self, value):
"""Add an element."""
raiseNotImplementedError

@abstractmethod
defdiscard(self, value):
"""Remove an element. Do not raise an exception if absent."""
raiseNotImplementedError

defremove(self, value):
"""Remove an element. If not a member, raise a KeyError."""
ifvaluenotinself:
raiseKeyError(value)
self.discard(value)

defpop(self):
"""Return the popped value. Raise KeyError if empty."""
it=iter(self)
try:
value=next(it)
exceptStopIteration:
raiseKeyErrorfromNone
self.discard(value)
returnvalue

defclear(self):
"""This is slow (creates N new iterators!) but effective."""
try:
whileTrue:
self.pop()
exceptKeyError:
pass

def__ior__(self, it):
forvalueinit:
self.add(value)
returnself

def__iand__(self, it):
forvaluein (self-it):
self.discard(value)
returnself

def__ixor__(self, it):
ifitisself:
self.clear()
else:
ifnotisinstance(it, Set):
it=self._from_iterable(it)
forvalueinit:
ifvalueinself:
self.discard(value)
else:
self.add(value)
returnself

def__isub__(self, it):
ifitisself:
self.clear()
else:
forvalueinit:
self.discard(value)
returnself


MutableSet.register(set)


### MAPPINGS ###

classMapping(Collection):
"""A Mapping is a generic container for associating key/value
 pairs.

 This class provides concrete generic implementations of all
 methods except for __getitem__, __iter__, and __len__.
 """

__slots__= ()

# Tell ABCMeta.__new__ that this class should have TPFLAGS_MAPPING set.
__abc_tpflags__=1<<6# Py_TPFLAGS_MAPPING

@abstractmethod
def__getitem__(self, key):
raiseKeyError

defget(self, key, default=None):
'D.get(k[,d]) -> D[k] if k in D, else d. d defaults to None.'
try:
returnself[key]
exceptKeyError:
returndefault

def__contains__(self, key):
try:
self[key]
exceptKeyError:
returnFalse
else:
returnTrue

defkeys(self):
"D.keys() -> a set-like object providing a view on D's keys"
returnKeysView(self)

defitems(self):
"D.items() -> a set-like object providing a view on D's items"
returnItemsView(self)

defvalues(self):
"D.values() -> an object providing a view on D's values"
returnValuesView(self)

def__eq__(self, other):
ifnotisinstance(other, Mapping):
returnNotImplemented
returndict(self.items()) ==dict(other.items())

__reversed__=None

Mapping.register(mappingproxy)
Mapping.register(framelocalsproxy)


classMappingView(Sized):

__slots__='_mapping',

def__init__(self, mapping):
self._mapping=mapping

def__len__(self):
returnlen(self._mapping)

def__repr__(self):
return'{0.__class__.__name__}({0._mapping!r})'.format(self)

__class_getitem__=classmethod(GenericAlias)


classKeysView(MappingView, Set):

__slots__= ()

@classmethod
def_from_iterable(cls, it):
returnset(it)

def__contains__(self, key):
returnkeyinself._mapping

def__iter__(self):
yieldfromself._mapping


KeysView.register(dict_keys)


classItemsView(MappingView, Set):

__slots__= ()

@classmethod
def_from_iterable(cls, it):
returnset(it)

def__contains__(self, item):
key, value=item
try:
v=self._mapping[key]
exceptKeyError:
returnFalse
else:
returnvisvalueorv==value

def__iter__(self):
forkeyinself._mapping:
yield (key, self._mapping[key])


ItemsView.register(dict_items)


classValuesView(MappingView, Collection):

__slots__= ()

def__contains__(self, value):
forkeyinself._mapping:
v=self._mapping[key]
ifvisvalueorv==value:
returnTrue
returnFalse

def__iter__(self):
forkeyinself._mapping:
yieldself._mapping[key]


ValuesView.register(dict_values)


classMutableMapping(Mapping):
"""A MutableMapping is a generic container for associating
 key/value pairs.

 This class provides concrete generic implementations of all
 methods except for __getitem__, __setitem__, __delitem__,
 __iter__, and __len__.
 """

__slots__= ()

@abstractmethod
def__setitem__(self, key, value):
raiseKeyError

@abstractmethod
def__delitem__(self, key):
raiseKeyError

__marker=object()

defpop(self, key, default=__marker):
'''D.pop(k[,d]) -> v, remove specified key and return the corresponding value.
 If key is not found, d is returned if given, otherwise KeyError is raised.
 '''
try:
value=self[key]
exceptKeyError:
ifdefaultisself.__marker:
raise
returndefault
else:
delself[key]
returnvalue

defpopitem(self):
'''D.popitem() -> (k, v), remove and return some (key, value) pair
 as a 2-tuple; but raise KeyError if D is empty.
 '''
try:
key=next(iter(self))
exceptStopIteration:
raiseKeyErrorfromNone
value=self[key]
delself[key]
returnkey, value

defclear(self):
'D.clear() -> None. Remove all items from D.'
try:
whileTrue:
self.popitem()
exceptKeyError:
pass

defupdate(self, other=(), /, **kwds):
''' D.update([E, ]**F) -> None. Update D from mapping/iterable E and F.
 If E present and has a .keys() method, does: for k in E.keys(): D[k] = E[k]
 If E present and lacks .keys() method, does: for (k, v) in E: D[k] = v
 In either case, this is followed by: for k, v in F.items(): D[k] = v
 '''
ifisinstance(other, Mapping):
forkeyinother:
self[key] =other[key]
elifhasattr(other, "keys"):
forkeyinother.keys():
self[key] =other[key]
else:
forkey, valueinother:
self[key] =value
forkey, valueinkwds.items():
self[key] =value

defsetdefault(self, key, default=None):
'D.setdefault(k[,d]) -> D.get(k,d), also set D[k]=d if k not in D'
try:
returnself[key]