# 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.
from abc import ABCMeta, abstractmethod
__all__ = ["Awaitable", "Coroutine",
"AsyncIterable", "AsyncIterator", "AsyncGenerator",
"Hashable", "Iterable", "Iterator", "Generator", "Reversible",
"Sized", "Container", "Callable", "Collection",
"Mapping", "MutableMapping",
"MappingView", "KeysView", "ItemsView", "ValuesView",
"Sequence", "MutableSequence",
# 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.
__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()))
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()))
dict_keys = type({}.keys())
dict_values = type({}.values())
dict_items = type({}.items())
mappingproxy = type(type.__dict__)
generator = type((lambda: (yield))())
_coro.close() # Prevent ResourceWarning
## asynchronous generator ##
async_generator = type(_ag)
def _check_methods(C, *methods):
if B.__dict__[method] is None:
class Hashable(metaclass=ABCMeta):
def __subclasshook__(cls, C):
return _check_methods(C, "__hash__")
class Awaitable(metaclass=ABCMeta):
def __subclasshook__(cls, C):
return _check_methods(C, "__await__")
class Coroutine(Awaitable):
"""Send a value into the coroutine.
Return next yielded value or raise StopIteration.
def throw(self, typ, val=None, tb=None):
"""Raise an exception in the coroutine.
Return next yielded value or raise StopIteration.
val = val.with_traceback(tb)
"""Raise GeneratorExit inside coroutine.
self.throw(GeneratorExit)
except (GeneratorExit, StopIteration):
raise RuntimeError("coroutine ignored GeneratorExit")
def __subclasshook__(cls, C):
return _check_methods(C, '__await__', 'send', 'throw', 'close')
Coroutine.register(coroutine)
class AsyncIterable(metaclass=ABCMeta):
def __subclasshook__(cls, C):
return _check_methods(C, "__aiter__")
class AsyncIterator(AsyncIterable):
async def __anext__(self):
"""Return the next item or raise StopAsyncIteration when exhausted."""
def __subclasshook__(cls, C):
return _check_methods(C, "__anext__", "__aiter__")
class AsyncGenerator(AsyncIterator):
async def __anext__(self):
"""Return the next item from the asynchronous generator.
When exhausted, raise StopAsyncIteration.
return await self.asend(None)
async def asend(self, value):
"""Send a value into the asynchronous generator.
Return next yielded value or raise StopAsyncIteration.
async def athrow(self, typ, val=None, tb=None):
"""Raise an exception in the asynchronous generator.
Return next yielded value or raise StopAsyncIteration.
val = val.with_traceback(tb)
"""Raise GeneratorExit inside coroutine.
await self.athrow(GeneratorExit)
except (GeneratorExit, StopAsyncIteration):
raise RuntimeError("asynchronous generator ignored GeneratorExit")
def __subclasshook__(cls, C):
if cls is AsyncGenerator:
return _check_methods(C, '__aiter__', '__anext__',
'asend', 'athrow', 'aclose')
AsyncGenerator.register(async_generator)
class Iterable(metaclass=ABCMeta):
def __subclasshook__(cls, C):
return _check_methods(C, "__iter__")
class Iterator(Iterable):
'Return the next item from the iterator. When exhausted, raise StopIteration'
def __subclasshook__(cls, C):
return _check_methods(C, '__iter__', '__next__')
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)
class Reversible(Iterable):
def __subclasshook__(cls, C):
return _check_methods(C, "__reversed__", "__iter__")
class Generator(Iterator):
"""Return the next item from the generator.
When exhausted, raise StopIteration.
"""Send a value into the generator.
Return next yielded value or raise StopIteration.
def throw(self, typ, val=None, tb=None):
"""Raise an exception in the generator.
Return next yielded value or raise StopIteration.
val = val.with_traceback(tb)
"""Raise GeneratorExit inside generator.
self.throw(GeneratorExit)
except (GeneratorExit, StopIteration):
raise RuntimeError("generator ignored GeneratorExit")
def __subclasshook__(cls, C):
return _check_methods(C, '__iter__', '__next__',
'send', 'throw', 'close')
Generator.register(generator)
class Sized(metaclass=ABCMeta):
def __subclasshook__(cls, C):
return _check_methods(C, "__len__")
class Container(metaclass=ABCMeta):
def __contains__(self, x):
def __subclasshook__(cls, C):
return _check_methods(C, "__contains__")
class Collection(Sized, Iterable, Container):
def __subclasshook__(cls, C):
return _check_methods(C, "__len__", "__iter__", "__contains__")
class Callable(metaclass=ABCMeta):
def __call__(self, *args, **kwds):
def __subclasshook__(cls, C):
return _check_methods(C, "__call__")
"""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.
if not isinstance(other, Set):
if len(self) > len(other):
if not isinstance(other, Set):
return len(self) < len(other) and self.__le__(other)
if not isinstance(other, Set):
return len(self) > len(other) and self.__ge__(other)
if not isinstance(other, Set):
if len(self) < len(other):
if not isinstance(other, Set):
return len(self) == len(other) and self.__le__(other)
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.
def __and__(self, other):
if not isinstance(other, Iterable):
return self._from_iterable(value for value in other if value in self)
def isdisjoint(self, other):
'Return True if two sets have a null intersection.'
if not isinstance(other, Iterable):
chain = (e for s in (self, other) for e in s)
return self._from_iterable(chain)
def __sub__(self, other):
if not isinstance(other, Set):
if not isinstance(other, Iterable):
other = self._from_iterable(other)