ghgfbnvb
This commit is contained in:
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
Binary file not shown.
File diff suppressed because it is too large
Load Diff
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
@@ -1,68 +0,0 @@
|
||||
import re
|
||||
import textwrap
|
||||
import email.message
|
||||
|
||||
from ._text import FoldedCase
|
||||
|
||||
|
||||
class Message(email.message.Message):
|
||||
multiple_use_keys = set(
|
||||
map(
|
||||
FoldedCase,
|
||||
[
|
||||
'Classifier',
|
||||
'Obsoletes-Dist',
|
||||
'Platform',
|
||||
'Project-URL',
|
||||
'Provides-Dist',
|
||||
'Provides-Extra',
|
||||
'Requires-Dist',
|
||||
'Requires-External',
|
||||
'Supported-Platform',
|
||||
'Dynamic',
|
||||
],
|
||||
)
|
||||
)
|
||||
"""
|
||||
Keys that may be indicated multiple times per PEP 566.
|
||||
"""
|
||||
|
||||
def __new__(cls, orig: email.message.Message):
|
||||
res = super().__new__(cls)
|
||||
vars(res).update(vars(orig))
|
||||
return res
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
self._headers = self._repair_headers()
|
||||
|
||||
# suppress spurious error from mypy
|
||||
def __iter__(self):
|
||||
return super().__iter__()
|
||||
|
||||
def _repair_headers(self):
|
||||
def redent(value):
|
||||
"Correct for RFC822 indentation"
|
||||
if not value or '\n' not in value:
|
||||
return value
|
||||
return textwrap.dedent(' ' * 8 + value)
|
||||
|
||||
headers = [(key, redent(value)) for key, value in vars(self)['_headers']]
|
||||
if self._payload:
|
||||
headers.append(('Description', self.get_payload()))
|
||||
return headers
|
||||
|
||||
@property
|
||||
def json(self):
|
||||
"""
|
||||
Convert PackageMetadata to a JSON-compatible format
|
||||
per PEP 0566.
|
||||
"""
|
||||
|
||||
def transform(key):
|
||||
value = self.get_all(key) if key in self.multiple_use_keys else self[key]
|
||||
if key == 'Keywords':
|
||||
value = re.split(r'\s+', value)
|
||||
tk = key.lower().replace('-', '_')
|
||||
return tk, value
|
||||
|
||||
return dict(map(transform, map(FoldedCase, self)))
|
||||
-30
@@ -1,30 +0,0 @@
|
||||
import collections
|
||||
|
||||
|
||||
# from jaraco.collections 3.3
|
||||
class FreezableDefaultDict(collections.defaultdict):
|
||||
"""
|
||||
Often it is desirable to prevent the mutation of
|
||||
a default dict after its initial construction, such
|
||||
as to prevent mutation during iteration.
|
||||
|
||||
>>> dd = FreezableDefaultDict(list)
|
||||
>>> dd[0].append('1')
|
||||
>>> dd.freeze()
|
||||
>>> dd[1]
|
||||
[]
|
||||
>>> len(dd)
|
||||
1
|
||||
"""
|
||||
|
||||
def __missing__(self, key):
|
||||
return getattr(self, '_frozen', super().__missing__)(key)
|
||||
|
||||
def freeze(self):
|
||||
self._frozen = lambda key: self.default_factory()
|
||||
|
||||
|
||||
class Pair(collections.namedtuple('Pair', 'name value')):
|
||||
@classmethod
|
||||
def parse(cls, text):
|
||||
return cls(*map(str.strip, text.split("=", 1)))
|
||||
@@ -1,71 +0,0 @@
|
||||
import sys
|
||||
import platform
|
||||
|
||||
|
||||
__all__ = ['install', 'NullFinder', 'Protocol']
|
||||
|
||||
|
||||
try:
|
||||
from typing import Protocol
|
||||
except ImportError: # pragma: no cover
|
||||
from ..typing_extensions import Protocol # type: ignore
|
||||
|
||||
|
||||
def install(cls):
|
||||
"""
|
||||
Class decorator for installation on sys.meta_path.
|
||||
|
||||
Adds the backport DistributionFinder to sys.meta_path and
|
||||
attempts to disable the finder functionality of the stdlib
|
||||
DistributionFinder.
|
||||
"""
|
||||
sys.meta_path.append(cls())
|
||||
disable_stdlib_finder()
|
||||
return cls
|
||||
|
||||
|
||||
def disable_stdlib_finder():
|
||||
"""
|
||||
Give the backport primacy for discovering path-based distributions
|
||||
by monkey-patching the stdlib O_O.
|
||||
|
||||
See #91 for more background for rationale on this sketchy
|
||||
behavior.
|
||||
"""
|
||||
|
||||
def matches(finder):
|
||||
return getattr(
|
||||
finder, '__module__', None
|
||||
) == '_frozen_importlib_external' and hasattr(finder, 'find_distributions')
|
||||
|
||||
for finder in filter(matches, sys.meta_path): # pragma: nocover
|
||||
del finder.find_distributions
|
||||
|
||||
|
||||
class NullFinder:
|
||||
"""
|
||||
A "Finder" (aka "MetaClassFinder") that never finds any modules,
|
||||
but may find distributions.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def find_spec(*args, **kwargs):
|
||||
return None
|
||||
|
||||
# In Python 2, the import system requires finders
|
||||
# to have a find_module() method, but this usage
|
||||
# is deprecated in Python 3 in favor of find_spec().
|
||||
# For the purposes of this finder (i.e. being present
|
||||
# on sys.meta_path but having no other import
|
||||
# system functionality), the two methods are identical.
|
||||
find_module = find_spec
|
||||
|
||||
|
||||
def pypy_partial(val):
|
||||
"""
|
||||
Adjust for variable stacklevel on partial under PyPy.
|
||||
|
||||
Workaround for #327.
|
||||
"""
|
||||
is_pypy = platform.python_implementation() == 'PyPy'
|
||||
return val + is_pypy
|
||||
-104
@@ -1,104 +0,0 @@
|
||||
import types
|
||||
import functools
|
||||
|
||||
|
||||
# from jaraco.functools 3.3
|
||||
def method_cache(method, cache_wrapper=None):
|
||||
"""
|
||||
Wrap lru_cache to support storing the cache data in the object instances.
|
||||
|
||||
Abstracts the common paradigm where the method explicitly saves an
|
||||
underscore-prefixed protected property on first call and returns that
|
||||
subsequently.
|
||||
|
||||
>>> class MyClass:
|
||||
... calls = 0
|
||||
...
|
||||
... @method_cache
|
||||
... def method(self, value):
|
||||
... self.calls += 1
|
||||
... return value
|
||||
|
||||
>>> a = MyClass()
|
||||
>>> a.method(3)
|
||||
3
|
||||
>>> for x in range(75):
|
||||
... res = a.method(x)
|
||||
>>> a.calls
|
||||
75
|
||||
|
||||
Note that the apparent behavior will be exactly like that of lru_cache
|
||||
except that the cache is stored on each instance, so values in one
|
||||
instance will not flush values from another, and when an instance is
|
||||
deleted, so are the cached values for that instance.
|
||||
|
||||
>>> b = MyClass()
|
||||
>>> for x in range(35):
|
||||
... res = b.method(x)
|
||||
>>> b.calls
|
||||
35
|
||||
>>> a.method(0)
|
||||
0
|
||||
>>> a.calls
|
||||
75
|
||||
|
||||
Note that if method had been decorated with ``functools.lru_cache()``,
|
||||
a.calls would have been 76 (due to the cached value of 0 having been
|
||||
flushed by the 'b' instance).
|
||||
|
||||
Clear the cache with ``.cache_clear()``
|
||||
|
||||
>>> a.method.cache_clear()
|
||||
|
||||
Same for a method that hasn't yet been called.
|
||||
|
||||
>>> c = MyClass()
|
||||
>>> c.method.cache_clear()
|
||||
|
||||
Another cache wrapper may be supplied:
|
||||
|
||||
>>> cache = functools.lru_cache(maxsize=2)
|
||||
>>> MyClass.method2 = method_cache(lambda self: 3, cache_wrapper=cache)
|
||||
>>> a = MyClass()
|
||||
>>> a.method2()
|
||||
3
|
||||
|
||||
Caution - do not subsequently wrap the method with another decorator, such
|
||||
as ``@property``, which changes the semantics of the function.
|
||||
|
||||
See also
|
||||
http://code.activestate.com/recipes/577452-a-memoize-decorator-for-instance-methods/
|
||||
for another implementation and additional justification.
|
||||
"""
|
||||
cache_wrapper = cache_wrapper or functools.lru_cache()
|
||||
|
||||
def wrapper(self, *args, **kwargs):
|
||||
# it's the first call, replace the method with a cached, bound method
|
||||
bound_method = types.MethodType(method, self)
|
||||
cached_method = cache_wrapper(bound_method)
|
||||
setattr(self, method.__name__, cached_method)
|
||||
return cached_method(*args, **kwargs)
|
||||
|
||||
# Support cache clear even before cache has been created.
|
||||
wrapper.cache_clear = lambda: None
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
# From jaraco.functools 3.3
|
||||
def pass_none(func):
|
||||
"""
|
||||
Wrap func so it's not called if its first param is None
|
||||
|
||||
>>> print_text = pass_none(print)
|
||||
>>> print_text('text')
|
||||
text
|
||||
>>> print_text(None)
|
||||
"""
|
||||
|
||||
@functools.wraps(func)
|
||||
def wrapper(param, *args, **kwargs):
|
||||
if param is not None:
|
||||
return func(param, *args, **kwargs)
|
||||
|
||||
return wrapper
|
||||
@@ -1,73 +0,0 @@
|
||||
from itertools import filterfalse
|
||||
|
||||
|
||||
def unique_everseen(iterable, key=None):
|
||||
"List unique elements, preserving order. Remember all elements ever seen."
|
||||
# unique_everseen('AAAABBBCCDAABBB') --> A B C D
|
||||
# unique_everseen('ABBCcAD', str.lower) --> A B C D
|
||||
seen = set()
|
||||
seen_add = seen.add
|
||||
if key is None:
|
||||
for element in filterfalse(seen.__contains__, iterable):
|
||||
seen_add(element)
|
||||
yield element
|
||||
else:
|
||||
for element in iterable:
|
||||
k = key(element)
|
||||
if k not in seen:
|
||||
seen_add(k)
|
||||
yield element
|
||||
|
||||
|
||||
# copied from more_itertools 8.8
|
||||
def always_iterable(obj, base_type=(str, bytes)):
|
||||
"""If *obj* is iterable, return an iterator over its items::
|
||||
|
||||
>>> obj = (1, 2, 3)
|
||||
>>> list(always_iterable(obj))
|
||||
[1, 2, 3]
|
||||
|
||||
If *obj* is not iterable, return a one-item iterable containing *obj*::
|
||||
|
||||
>>> obj = 1
|
||||
>>> list(always_iterable(obj))
|
||||
[1]
|
||||
|
||||
If *obj* is ``None``, return an empty iterable:
|
||||
|
||||
>>> obj = None
|
||||
>>> list(always_iterable(None))
|
||||
[]
|
||||
|
||||
By default, binary and text strings are not considered iterable::
|
||||
|
||||
>>> obj = 'foo'
|
||||
>>> list(always_iterable(obj))
|
||||
['foo']
|
||||
|
||||
If *base_type* is set, objects for which ``isinstance(obj, base_type)``
|
||||
returns ``True`` won't be considered iterable.
|
||||
|
||||
>>> obj = {'a': 1}
|
||||
>>> list(always_iterable(obj)) # Iterate over the dict's keys
|
||||
['a']
|
||||
>>> list(always_iterable(obj, base_type=dict)) # Treat dicts as a unit
|
||||
[{'a': 1}]
|
||||
|
||||
Set *base_type* to ``None`` to avoid any special handling and treat objects
|
||||
Python considers iterable as iterable:
|
||||
|
||||
>>> obj = 'foo'
|
||||
>>> list(always_iterable(obj, base_type=None))
|
||||
['f', 'o', 'o']
|
||||
"""
|
||||
if obj is None:
|
||||
return iter(())
|
||||
|
||||
if (base_type is not None) and isinstance(obj, base_type):
|
||||
return iter((obj,))
|
||||
|
||||
try:
|
||||
return iter(obj)
|
||||
except TypeError:
|
||||
return iter((obj,))
|
||||
@@ -1,48 +0,0 @@
|
||||
from ._compat import Protocol
|
||||
from typing import Any, Dict, Iterator, List, TypeVar, Union
|
||||
|
||||
|
||||
_T = TypeVar("_T")
|
||||
|
||||
|
||||
class PackageMetadata(Protocol):
|
||||
def __len__(self) -> int:
|
||||
... # pragma: no cover
|
||||
|
||||
def __contains__(self, item: str) -> bool:
|
||||
... # pragma: no cover
|
||||
|
||||
def __getitem__(self, key: str) -> str:
|
||||
... # pragma: no cover
|
||||
|
||||
def __iter__(self) -> Iterator[str]:
|
||||
... # pragma: no cover
|
||||
|
||||
def get_all(self, name: str, failobj: _T = ...) -> Union[List[Any], _T]:
|
||||
"""
|
||||
Return all values associated with a possibly multi-valued key.
|
||||
"""
|
||||
|
||||
@property
|
||||
def json(self) -> Dict[str, Union[str, List[str]]]:
|
||||
"""
|
||||
A JSON-compatible form of the metadata.
|
||||
"""
|
||||
|
||||
|
||||
class SimplePath(Protocol):
|
||||
"""
|
||||
A minimal subset of pathlib.Path required by PathDistribution.
|
||||
"""
|
||||
|
||||
def joinpath(self) -> 'SimplePath':
|
||||
... # pragma: no cover
|
||||
|
||||
def __truediv__(self) -> 'SimplePath':
|
||||
... # pragma: no cover
|
||||
|
||||
def parent(self) -> 'SimplePath':
|
||||
... # pragma: no cover
|
||||
|
||||
def read_text(self) -> str:
|
||||
... # pragma: no cover
|
||||
@@ -1,99 +0,0 @@
|
||||
import re
|
||||
|
||||
from ._functools import method_cache
|
||||
|
||||
|
||||
# from jaraco.text 3.5
|
||||
class FoldedCase(str):
|
||||
"""
|
||||
A case insensitive string class; behaves just like str
|
||||
except compares equal when the only variation is case.
|
||||
|
||||
>>> s = FoldedCase('hello world')
|
||||
|
||||
>>> s == 'Hello World'
|
||||
True
|
||||
|
||||
>>> 'Hello World' == s
|
||||
True
|
||||
|
||||
>>> s != 'Hello World'
|
||||
False
|
||||
|
||||
>>> s.index('O')
|
||||
4
|
||||
|
||||
>>> s.split('O')
|
||||
['hell', ' w', 'rld']
|
||||
|
||||
>>> sorted(map(FoldedCase, ['GAMMA', 'alpha', 'Beta']))
|
||||
['alpha', 'Beta', 'GAMMA']
|
||||
|
||||
Sequence membership is straightforward.
|
||||
|
||||
>>> "Hello World" in [s]
|
||||
True
|
||||
>>> s in ["Hello World"]
|
||||
True
|
||||
|
||||
You may test for set inclusion, but candidate and elements
|
||||
must both be folded.
|
||||
|
||||
>>> FoldedCase("Hello World") in {s}
|
||||
True
|
||||
>>> s in {FoldedCase("Hello World")}
|
||||
True
|
||||
|
||||
String inclusion works as long as the FoldedCase object
|
||||
is on the right.
|
||||
|
||||
>>> "hello" in FoldedCase("Hello World")
|
||||
True
|
||||
|
||||
But not if the FoldedCase object is on the left:
|
||||
|
||||
>>> FoldedCase('hello') in 'Hello World'
|
||||
False
|
||||
|
||||
In that case, use in_:
|
||||
|
||||
>>> FoldedCase('hello').in_('Hello World')
|
||||
True
|
||||
|
||||
>>> FoldedCase('hello') > FoldedCase('Hello')
|
||||
False
|
||||
"""
|
||||
|
||||
def __lt__(self, other):
|
||||
return self.lower() < other.lower()
|
||||
|
||||
def __gt__(self, other):
|
||||
return self.lower() > other.lower()
|
||||
|
||||
def __eq__(self, other):
|
||||
return self.lower() == other.lower()
|
||||
|
||||
def __ne__(self, other):
|
||||
return self.lower() != other.lower()
|
||||
|
||||
def __hash__(self):
|
||||
return hash(self.lower())
|
||||
|
||||
def __contains__(self, other):
|
||||
return super().lower().__contains__(other.lower())
|
||||
|
||||
def in_(self, other):
|
||||
"Does self appear in other?"
|
||||
return self in FoldedCase(other)
|
||||
|
||||
# cache lower since it's likely to be called frequently.
|
||||
@method_cache
|
||||
def lower(self):
|
||||
return super().lower()
|
||||
|
||||
def index(self, sub):
|
||||
return self.lower().index(sub.lower())
|
||||
|
||||
def split(self, splitter=' ', maxsplit=0):
|
||||
pattern = re.compile(re.escape(splitter), re.I)
|
||||
return pattern.split(self, maxsplit)
|
||||
@@ -1,36 +0,0 @@
|
||||
"""Read resources contained within a package."""
|
||||
|
||||
from ._common import (
|
||||
as_file,
|
||||
files,
|
||||
Package,
|
||||
)
|
||||
|
||||
from ._legacy import (
|
||||
contents,
|
||||
open_binary,
|
||||
read_binary,
|
||||
open_text,
|
||||
read_text,
|
||||
is_resource,
|
||||
path,
|
||||
Resource,
|
||||
)
|
||||
|
||||
from .abc import ResourceReader
|
||||
|
||||
|
||||
__all__ = [
|
||||
'Package',
|
||||
'Resource',
|
||||
'ResourceReader',
|
||||
'as_file',
|
||||
'contents',
|
||||
'files',
|
||||
'is_resource',
|
||||
'open_binary',
|
||||
'open_text',
|
||||
'path',
|
||||
'read_binary',
|
||||
'read_text',
|
||||
]
|
||||
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
-170
@@ -1,170 +0,0 @@
|
||||
from contextlib import suppress
|
||||
from io import TextIOWrapper
|
||||
|
||||
from . import abc
|
||||
|
||||
|
||||
class SpecLoaderAdapter:
|
||||
"""
|
||||
Adapt a package spec to adapt the underlying loader.
|
||||
"""
|
||||
|
||||
def __init__(self, spec, adapter=lambda spec: spec.loader):
|
||||
self.spec = spec
|
||||
self.loader = adapter(spec)
|
||||
|
||||
def __getattr__(self, name):
|
||||
return getattr(self.spec, name)
|
||||
|
||||
|
||||
class TraversableResourcesLoader:
|
||||
"""
|
||||
Adapt a loader to provide TraversableResources.
|
||||
"""
|
||||
|
||||
def __init__(self, spec):
|
||||
self.spec = spec
|
||||
|
||||
def get_resource_reader(self, name):
|
||||
return CompatibilityFiles(self.spec)._native()
|
||||
|
||||
|
||||
def _io_wrapper(file, mode='r', *args, **kwargs):
|
||||
if mode == 'r':
|
||||
return TextIOWrapper(file, *args, **kwargs)
|
||||
elif mode == 'rb':
|
||||
return file
|
||||
raise ValueError(
|
||||
"Invalid mode value '{}', only 'r' and 'rb' are supported".format(mode)
|
||||
)
|
||||
|
||||
|
||||
class CompatibilityFiles:
|
||||
"""
|
||||
Adapter for an existing or non-existent resource reader
|
||||
to provide a compatibility .files().
|
||||
"""
|
||||
|
||||
class SpecPath(abc.Traversable):
|
||||
"""
|
||||
Path tied to a module spec.
|
||||
Can be read and exposes the resource reader children.
|
||||
"""
|
||||
|
||||
def __init__(self, spec, reader):
|
||||
self._spec = spec
|
||||
self._reader = reader
|
||||
|
||||
def iterdir(self):
|
||||
if not self._reader:
|
||||
return iter(())
|
||||
return iter(
|
||||
CompatibilityFiles.ChildPath(self._reader, path)
|
||||
for path in self._reader.contents()
|
||||
)
|
||||
|
||||
def is_file(self):
|
||||
return False
|
||||
|
||||
is_dir = is_file
|
||||
|
||||
def joinpath(self, other):
|
||||
if not self._reader:
|
||||
return CompatibilityFiles.OrphanPath(other)
|
||||
return CompatibilityFiles.ChildPath(self._reader, other)
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
return self._spec.name
|
||||
|
||||
def open(self, mode='r', *args, **kwargs):
|
||||
return _io_wrapper(self._reader.open_resource(None), mode, *args, **kwargs)
|
||||
|
||||
class ChildPath(abc.Traversable):
|
||||
"""
|
||||
Path tied to a resource reader child.
|
||||
Can be read but doesn't expose any meaningful children.
|
||||
"""
|
||||
|
||||
def __init__(self, reader, name):
|
||||
self._reader = reader
|
||||
self._name = name
|
||||
|
||||
def iterdir(self):
|
||||
return iter(())
|
||||
|
||||
def is_file(self):
|
||||
return self._reader.is_resource(self.name)
|
||||
|
||||
def is_dir(self):
|
||||
return not self.is_file()
|
||||
|
||||
def joinpath(self, other):
|
||||
return CompatibilityFiles.OrphanPath(self.name, other)
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
return self._name
|
||||
|
||||
def open(self, mode='r', *args, **kwargs):
|
||||
return _io_wrapper(
|
||||
self._reader.open_resource(self.name), mode, *args, **kwargs
|
||||
)
|
||||
|
||||
class OrphanPath(abc.Traversable):
|
||||
"""
|
||||
Orphan path, not tied to a module spec or resource reader.
|
||||
Can't be read and doesn't expose any meaningful children.
|
||||
"""
|
||||
|
||||
def __init__(self, *path_parts):
|
||||
if len(path_parts) < 1:
|
||||
raise ValueError('Need at least one path part to construct a path')
|
||||
self._path = path_parts
|
||||
|
||||
def iterdir(self):
|
||||
return iter(())
|
||||
|
||||
def is_file(self):
|
||||
return False
|
||||
|
||||
is_dir = is_file
|
||||
|
||||
def joinpath(self, other):
|
||||
return CompatibilityFiles.OrphanPath(*self._path, other)
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
return self._path[-1]
|
||||
|
||||
def open(self, mode='r', *args, **kwargs):
|
||||
raise FileNotFoundError("Can't open orphan path")
|
||||
|
||||
def __init__(self, spec):
|
||||
self.spec = spec
|
||||
|
||||
@property
|
||||
def _reader(self):
|
||||
with suppress(AttributeError):
|
||||
return self.spec.loader.get_resource_reader(self.spec.name)
|
||||
|
||||
def _native(self):
|
||||
"""
|
||||
Return the native reader if it supports files().
|
||||
"""
|
||||
reader = self._reader
|
||||
return reader if hasattr(reader, 'files') else self
|
||||
|
||||
def __getattr__(self, attr):
|
||||
return getattr(self._reader, attr)
|
||||
|
||||
def files(self):
|
||||
return CompatibilityFiles.SpecPath(self.spec, self._reader)
|
||||
|
||||
|
||||
def wrap_spec(package):
|
||||
"""
|
||||
Construct a package spec with traversable compatibility
|
||||
on the spec/loader/reader.
|
||||
"""
|
||||
return SpecLoaderAdapter(package.__spec__, TraversableResourcesLoader)
|
||||
@@ -1,104 +0,0 @@
|
||||
import os
|
||||
import pathlib
|
||||
import tempfile
|
||||
import functools
|
||||
import contextlib
|
||||
import types
|
||||
import importlib
|
||||
|
||||
from typing import Union, Optional
|
||||
from .abc import ResourceReader, Traversable
|
||||
|
||||
from ._compat import wrap_spec
|
||||
|
||||
Package = Union[types.ModuleType, str]
|
||||
|
||||
|
||||
def files(package):
|
||||
# type: (Package) -> Traversable
|
||||
"""
|
||||
Get a Traversable resource from a package
|
||||
"""
|
||||
return from_package(get_package(package))
|
||||
|
||||
|
||||
def get_resource_reader(package):
|
||||
# type: (types.ModuleType) -> Optional[ResourceReader]
|
||||
"""
|
||||
Return the package's loader if it's a ResourceReader.
|
||||
"""
|
||||
# We can't use
|
||||
# a issubclass() check here because apparently abc.'s __subclasscheck__()
|
||||
# hook wants to create a weak reference to the object, but
|
||||
# zipimport.zipimporter does not support weak references, resulting in a
|
||||
# TypeError. That seems terrible.
|
||||
spec = package.__spec__
|
||||
reader = getattr(spec.loader, 'get_resource_reader', None) # type: ignore
|
||||
if reader is None:
|
||||
return None
|
||||
return reader(spec.name) # type: ignore
|
||||
|
||||
|
||||
def resolve(cand):
|
||||
# type: (Package) -> types.ModuleType
|
||||
return cand if isinstance(cand, types.ModuleType) else importlib.import_module(cand)
|
||||
|
||||
|
||||
def get_package(package):
|
||||
# type: (Package) -> types.ModuleType
|
||||
"""Take a package name or module object and return the module.
|
||||
|
||||
Raise an exception if the resolved module is not a package.
|
||||
"""
|
||||
resolved = resolve(package)
|
||||
if wrap_spec(resolved).submodule_search_locations is None:
|
||||
raise TypeError(f'{package!r} is not a package')
|
||||
return resolved
|
||||
|
||||
|
||||
def from_package(package):
|
||||
"""
|
||||
Return a Traversable object for the given package.
|
||||
|
||||
"""
|
||||
spec = wrap_spec(package)
|
||||
reader = spec.loader.get_resource_reader(spec.name)
|
||||
return reader.files()
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _tempfile(reader, suffix=''):
|
||||
# Not using tempfile.NamedTemporaryFile as it leads to deeper 'try'
|
||||
# blocks due to the need to close the temporary file to work on Windows
|
||||
# properly.
|
||||
fd, raw_path = tempfile.mkstemp(suffix=suffix)
|
||||
try:
|
||||
try:
|
||||
os.write(fd, reader())
|
||||
finally:
|
||||
os.close(fd)
|
||||
del reader
|
||||
yield pathlib.Path(raw_path)
|
||||
finally:
|
||||
try:
|
||||
os.remove(raw_path)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
||||
|
||||
@functools.singledispatch
|
||||
def as_file(path):
|
||||
"""
|
||||
Given a Traversable object, return that object as a
|
||||
path on the local file system in a context manager.
|
||||
"""
|
||||
return _tempfile(path.read_bytes, suffix=path.name)
|
||||
|
||||
|
||||
@as_file.register(pathlib.Path)
|
||||
@contextlib.contextmanager
|
||||
def _(path):
|
||||
"""
|
||||
Degenerate behavior for pathlib.Path objects.
|
||||
"""
|
||||
yield path
|
||||
@@ -1,98 +0,0 @@
|
||||
# flake8: noqa
|
||||
|
||||
import abc
|
||||
import sys
|
||||
import pathlib
|
||||
from contextlib import suppress
|
||||
|
||||
if sys.version_info >= (3, 10):
|
||||
from zipfile import Path as ZipPath # type: ignore
|
||||
else:
|
||||
from ..zipp import Path as ZipPath # type: ignore
|
||||
|
||||
|
||||
try:
|
||||
from typing import runtime_checkable # type: ignore
|
||||
except ImportError:
|
||||
|
||||
def runtime_checkable(cls): # type: ignore
|
||||
return cls
|
||||
|
||||
|
||||
try:
|
||||
from typing import Protocol # type: ignore
|
||||
except ImportError:
|
||||
Protocol = abc.ABC # type: ignore
|
||||
|
||||
|
||||
class TraversableResourcesLoader:
|
||||
"""
|
||||
Adapt loaders to provide TraversableResources and other
|
||||
compatibility.
|
||||
|
||||
Used primarily for Python 3.9 and earlier where the native
|
||||
loaders do not yet implement TraversableResources.
|
||||
"""
|
||||
|
||||
def __init__(self, spec):
|
||||
self.spec = spec
|
||||
|
||||
@property
|
||||
def path(self):
|
||||
return self.spec.origin
|
||||
|
||||
def get_resource_reader(self, name):
|
||||
from . import readers, _adapters
|
||||
|
||||
def _zip_reader(spec):
|
||||
with suppress(AttributeError):
|
||||
return readers.ZipReader(spec.loader, spec.name)
|
||||
|
||||
def _namespace_reader(spec):
|
||||
with suppress(AttributeError, ValueError):
|
||||
return readers.NamespaceReader(spec.submodule_search_locations)
|
||||
|
||||
def _available_reader(spec):
|
||||
with suppress(AttributeError):
|
||||
return spec.loader.get_resource_reader(spec.name)
|
||||
|
||||
def _native_reader(spec):
|
||||
reader = _available_reader(spec)
|
||||
return reader if hasattr(reader, 'files') else None
|
||||
|
||||
def _file_reader(spec):
|
||||
try:
|
||||
path = pathlib.Path(self.path)
|
||||
except TypeError:
|
||||
return None
|
||||
if path.exists():
|
||||
return readers.FileReader(self)
|
||||
|
||||
return (
|
||||
# native reader if it supplies 'files'
|
||||
_native_reader(self.spec)
|
||||
or
|
||||
# local ZipReader if a zip module
|
||||
_zip_reader(self.spec)
|
||||
or
|
||||
# local NamespaceReader if a namespace module
|
||||
_namespace_reader(self.spec)
|
||||
or
|
||||
# local FileReader
|
||||
_file_reader(self.spec)
|
||||
# fallback - adapt the spec ResourceReader to TraversableReader
|
||||
or _adapters.CompatibilityFiles(self.spec)
|
||||
)
|
||||
|
||||
|
||||
def wrap_spec(package):
|
||||
"""
|
||||
Construct a package spec with traversable compatibility
|
||||
on the spec/loader/reader.
|
||||
|
||||
Supersedes _adapters.wrap_spec to use TraversableResourcesLoader
|
||||
from above for older Python compatibility (<3.10).
|
||||
"""
|
||||
from . import _adapters
|
||||
|
||||
return _adapters.SpecLoaderAdapter(package.__spec__, TraversableResourcesLoader)
|
||||
-35
@@ -1,35 +0,0 @@
|
||||
from itertools import filterfalse
|
||||
|
||||
from typing import (
|
||||
Callable,
|
||||
Iterable,
|
||||
Iterator,
|
||||
Optional,
|
||||
Set,
|
||||
TypeVar,
|
||||
Union,
|
||||
)
|
||||
|
||||
# Type and type variable definitions
|
||||
_T = TypeVar('_T')
|
||||
_U = TypeVar('_U')
|
||||
|
||||
|
||||
def unique_everseen(
|
||||
iterable: Iterable[_T], key: Optional[Callable[[_T], _U]] = None
|
||||
) -> Iterator[_T]:
|
||||
"List unique elements, preserving order. Remember all elements ever seen."
|
||||
# unique_everseen('AAAABBBCCDAABBB') --> A B C D
|
||||
# unique_everseen('ABBCcAD', str.lower) --> A B C D
|
||||
seen: Set[Union[_T, _U]] = set()
|
||||
seen_add = seen.add
|
||||
if key is None:
|
||||
for element in filterfalse(seen.__contains__, iterable):
|
||||
seen_add(element)
|
||||
yield element
|
||||
else:
|
||||
for element in iterable:
|
||||
k = key(element)
|
||||
if k not in seen:
|
||||
seen_add(k)
|
||||
yield element
|
||||
@@ -1,121 +0,0 @@
|
||||
import functools
|
||||
import os
|
||||
import pathlib
|
||||
import types
|
||||
import warnings
|
||||
|
||||
from typing import Union, Iterable, ContextManager, BinaryIO, TextIO, Any
|
||||
|
||||
from . import _common
|
||||
|
||||
Package = Union[types.ModuleType, str]
|
||||
Resource = str
|
||||
|
||||
|
||||
def deprecated(func):
|
||||
@functools.wraps(func)
|
||||
def wrapper(*args, **kwargs):
|
||||
warnings.warn(
|
||||
f"{func.__name__} is deprecated. Use files() instead. "
|
||||
"Refer to https://importlib-resources.readthedocs.io"
|
||||
"/en/latest/using.html#migrating-from-legacy for migration advice.",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
return func(*args, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
def normalize_path(path):
|
||||
# type: (Any) -> str
|
||||
"""Normalize a path by ensuring it is a string.
|
||||
|
||||
If the resulting string contains path separators, an exception is raised.
|
||||
"""
|
||||
str_path = str(path)
|
||||
parent, file_name = os.path.split(str_path)
|
||||
if parent:
|
||||
raise ValueError(f'{path!r} must be only a file name')
|
||||
return file_name
|
||||
|
||||
|
||||
@deprecated
|
||||
def open_binary(package: Package, resource: Resource) -> BinaryIO:
|
||||
"""Return a file-like object opened for binary reading of the resource."""
|
||||
return (_common.files(package) / normalize_path(resource)).open('rb')
|
||||
|
||||
|
||||
@deprecated
|
||||
def read_binary(package: Package, resource: Resource) -> bytes:
|
||||
"""Return the binary contents of the resource."""
|
||||
return (_common.files(package) / normalize_path(resource)).read_bytes()
|
||||
|
||||
|
||||
@deprecated
|
||||
def open_text(
|
||||
package: Package,
|
||||
resource: Resource,
|
||||
encoding: str = 'utf-8',
|
||||
errors: str = 'strict',
|
||||
) -> TextIO:
|
||||
"""Return a file-like object opened for text reading of the resource."""
|
||||
return (_common.files(package) / normalize_path(resource)).open(
|
||||
'r', encoding=encoding, errors=errors
|
||||
)
|
||||
|
||||
|
||||
@deprecated
|
||||
def read_text(
|
||||
package: Package,
|
||||
resource: Resource,
|
||||
encoding: str = 'utf-8',
|
||||
errors: str = 'strict',
|
||||
) -> str:
|
||||
"""Return the decoded string of the resource.
|
||||
|
||||
The decoding-related arguments have the same semantics as those of
|
||||
bytes.decode().
|
||||
"""
|
||||
with open_text(package, resource, encoding, errors) as fp:
|
||||
return fp.read()
|
||||
|
||||
|
||||
@deprecated
|
||||
def contents(package: Package) -> Iterable[str]:
|
||||
"""Return an iterable of entries in `package`.
|
||||
|
||||
Note that not all entries are resources. Specifically, directories are
|
||||
not considered resources. Use `is_resource()` on each entry returned here
|
||||
to check if it is a resource or not.
|
||||
"""
|
||||
return [path.name for path in _common.files(package).iterdir()]
|
||||
|
||||
|
||||
@deprecated
|
||||
def is_resource(package: Package, name: str) -> bool:
|
||||
"""True if `name` is a resource inside `package`.
|
||||
|
||||
Directories are *not* resources.
|
||||
"""
|
||||
resource = normalize_path(name)
|
||||
return any(
|
||||
traversable.name == resource and traversable.is_file()
|
||||
for traversable in _common.files(package).iterdir()
|
||||
)
|
||||
|
||||
|
||||
@deprecated
|
||||
def path(
|
||||
package: Package,
|
||||
resource: Resource,
|
||||
) -> ContextManager[pathlib.Path]:
|
||||
"""A context manager providing a file path object to the resource.
|
||||
|
||||
If the resource does not already exist on its own on the file system,
|
||||
a temporary file will be created. If the file was created, the file
|
||||
will be deleted upon exiting the context manager (no exception is
|
||||
raised if the file was deleted prior to the context manager
|
||||
exiting).
|
||||
"""
|
||||
return _common.as_file(_common.files(package) / normalize_path(resource))
|
||||
@@ -1,137 +0,0 @@
|
||||
import abc
|
||||
from typing import BinaryIO, Iterable, Text
|
||||
|
||||
from ._compat import runtime_checkable, Protocol
|
||||
|
||||
|
||||
class ResourceReader(metaclass=abc.ABCMeta):
|
||||
"""Abstract base class for loaders to provide resource reading support."""
|
||||
|
||||
@abc.abstractmethod
|
||||
def open_resource(self, resource: Text) -> BinaryIO:
|
||||
"""Return an opened, file-like object for binary reading.
|
||||
|
||||
The 'resource' argument is expected to represent only a file name.
|
||||
If the resource cannot be found, FileNotFoundError is raised.
|
||||
"""
|
||||
# This deliberately raises FileNotFoundError instead of
|
||||
# NotImplementedError so that if this method is accidentally called,
|
||||
# it'll still do the right thing.
|
||||
raise FileNotFoundError
|
||||
|
||||
@abc.abstractmethod
|
||||
def resource_path(self, resource: Text) -> Text:
|
||||
"""Return the file system path to the specified resource.
|
||||
|
||||
The 'resource' argument is expected to represent only a file name.
|
||||
If the resource does not exist on the file system, raise
|
||||
FileNotFoundError.
|
||||
"""
|
||||
# This deliberately raises FileNotFoundError instead of
|
||||
# NotImplementedError so that if this method is accidentally called,
|
||||
# it'll still do the right thing.
|
||||
raise FileNotFoundError
|
||||
|
||||
@abc.abstractmethod
|
||||
def is_resource(self, path: Text) -> bool:
|
||||
"""Return True if the named 'path' is a resource.
|
||||
|
||||
Files are resources, directories are not.
|
||||
"""
|
||||
raise FileNotFoundError
|
||||
|
||||
@abc.abstractmethod
|
||||
def contents(self) -> Iterable[str]:
|
||||
"""Return an iterable of entries in `package`."""
|
||||
raise FileNotFoundError
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class Traversable(Protocol):
|
||||
"""
|
||||
An object with a subset of pathlib.Path methods suitable for
|
||||
traversing directories and opening files.
|
||||
"""
|
||||
|
||||
@abc.abstractmethod
|
||||
def iterdir(self):
|
||||
"""
|
||||
Yield Traversable objects in self
|
||||
"""
|
||||
|
||||
def read_bytes(self):
|
||||
"""
|
||||
Read contents of self as bytes
|
||||
"""
|
||||
with self.open('rb') as strm:
|
||||
return strm.read()
|
||||
|
||||
def read_text(self, encoding=None):
|
||||
"""
|
||||
Read contents of self as text
|
||||
"""
|
||||
with self.open(encoding=encoding) as strm:
|
||||
return strm.read()
|
||||
|
||||
@abc.abstractmethod
|
||||
def is_dir(self) -> bool:
|
||||
"""
|
||||
Return True if self is a directory
|
||||
"""
|
||||
|
||||
@abc.abstractmethod
|
||||
def is_file(self) -> bool:
|
||||
"""
|
||||
Return True if self is a file
|
||||
"""
|
||||
|
||||
@abc.abstractmethod
|
||||
def joinpath(self, child):
|
||||
"""
|
||||
Return Traversable child in self
|
||||
"""
|
||||
|
||||
def __truediv__(self, child):
|
||||
"""
|
||||
Return Traversable child in self
|
||||
"""
|
||||
return self.joinpath(child)
|
||||
|
||||
@abc.abstractmethod
|
||||
def open(self, mode='r', *args, **kwargs):
|
||||
"""
|
||||
mode may be 'r' or 'rb' to open as text or binary. Return a handle
|
||||
suitable for reading (same as pathlib.Path.open).
|
||||
|
||||
When opening as text, accepts encoding parameters such as those
|
||||
accepted by io.TextIOWrapper.
|
||||
"""
|
||||
|
||||
@abc.abstractproperty
|
||||
def name(self) -> str:
|
||||
"""
|
||||
The base name of this object without any parent references.
|
||||
"""
|
||||
|
||||
|
||||
class TraversableResources(ResourceReader):
|
||||
"""
|
||||
The required interface for providing traversable
|
||||
resources.
|
||||
"""
|
||||
|
||||
@abc.abstractmethod
|
||||
def files(self):
|
||||
"""Return a Traversable object for the loaded package."""
|
||||
|
||||
def open_resource(self, resource):
|
||||
return self.files().joinpath(resource).open('rb')
|
||||
|
||||
def resource_path(self, resource):
|
||||
raise FileNotFoundError(resource)
|
||||
|
||||
def is_resource(self, path):
|
||||
return self.files().joinpath(path).is_file()
|
||||
|
||||
def contents(self):
|
||||
return (item.name for item in self.files().iterdir())
|
||||
@@ -1,122 +0,0 @@
|
||||
import collections
|
||||
import pathlib
|
||||
import operator
|
||||
|
||||
from . import abc
|
||||
|
||||
from ._itertools import unique_everseen
|
||||
from ._compat import ZipPath
|
||||
|
||||
|
||||
def remove_duplicates(items):
|
||||
return iter(collections.OrderedDict.fromkeys(items))
|
||||
|
||||
|
||||
class FileReader(abc.TraversableResources):
|
||||
def __init__(self, loader):
|
||||
self.path = pathlib.Path(loader.path).parent
|
||||
|
||||
def resource_path(self, resource):
|
||||
"""
|
||||
Return the file system path to prevent
|
||||
`resources.path()` from creating a temporary
|
||||
copy.
|
||||
"""
|
||||
return str(self.path.joinpath(resource))
|
||||
|
||||
def files(self):
|
||||
return self.path
|
||||
|
||||
|
||||
class ZipReader(abc.TraversableResources):
|
||||
def __init__(self, loader, module):
|
||||
_, _, name = module.rpartition('.')
|
||||
self.prefix = loader.prefix.replace('\\', '/') + name + '/'
|
||||
self.archive = loader.archive
|
||||
|
||||
def open_resource(self, resource):
|
||||
try:
|
||||
return super().open_resource(resource)
|
||||
except KeyError as exc:
|
||||
raise FileNotFoundError(exc.args[0])
|
||||
|
||||
def is_resource(self, path):
|
||||
# workaround for `zipfile.Path.is_file` returning true
|
||||
# for non-existent paths.
|
||||
target = self.files().joinpath(path)
|
||||
return target.is_file() and target.exists()
|
||||
|
||||
def files(self):
|
||||
return ZipPath(self.archive, self.prefix)
|
||||
|
||||
|
||||
class MultiplexedPath(abc.Traversable):
|
||||
"""
|
||||
Given a series of Traversable objects, implement a merged
|
||||
version of the interface across all objects. Useful for
|
||||
namespace packages which may be multihomed at a single
|
||||
name.
|
||||
"""
|
||||
|
||||
def __init__(self, *paths):
|
||||
self._paths = list(map(pathlib.Path, remove_duplicates(paths)))
|
||||
if not self._paths:
|
||||
message = 'MultiplexedPath must contain at least one path'
|
||||
raise FileNotFoundError(message)
|
||||
if not all(path.is_dir() for path in self._paths):
|
||||
raise NotADirectoryError('MultiplexedPath only supports directories')
|
||||
|
||||
def iterdir(self):
|
||||
files = (file for path in self._paths for file in path.iterdir())
|
||||
return unique_everseen(files, key=operator.attrgetter('name'))
|
||||
|
||||
def read_bytes(self):
|
||||
raise FileNotFoundError(f'{self} is not a file')
|
||||
|
||||
def read_text(self, *args, **kwargs):
|
||||
raise FileNotFoundError(f'{self} is not a file')
|
||||
|
||||
def is_dir(self):
|
||||
return True
|
||||
|
||||
def is_file(self):
|
||||
return False
|
||||
|
||||
def joinpath(self, child):
|
||||
# first try to find child in current paths
|
||||
for file in self.iterdir():
|
||||
if file.name == child:
|
||||
return file
|
||||
# if it does not exist, construct it with the first path
|
||||
return self._paths[0] / child
|
||||
|
||||
__truediv__ = joinpath
|
||||
|
||||
def open(self, *args, **kwargs):
|
||||
raise FileNotFoundError(f'{self} is not a file')
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
return self._paths[0].name
|
||||
|
||||
def __repr__(self):
|
||||
paths = ', '.join(f"'{path}'" for path in self._paths)
|
||||
return f'MultiplexedPath({paths})'
|
||||
|
||||
|
||||
class NamespaceReader(abc.TraversableResources):
|
||||
def __init__(self, namespace_path):
|
||||
if 'NamespacePath' not in str(namespace_path):
|
||||
raise ValueError('Invalid path')
|
||||
self.path = MultiplexedPath(*list(namespace_path))
|
||||
|
||||
def resource_path(self, resource):
|
||||
"""
|
||||
Return the file system path to prevent
|
||||
`resources.path()` from creating a temporary
|
||||
copy.
|
||||
"""
|
||||
return str(self.path.joinpath(resource))
|
||||
|
||||
def files(self):
|
||||
return self.path
|
||||
@@ -1,116 +0,0 @@
|
||||
"""
|
||||
Interface adapters for low-level readers.
|
||||
"""
|
||||
|
||||
import abc
|
||||
import io
|
||||
import itertools
|
||||
from typing import BinaryIO, List
|
||||
|
||||
from .abc import Traversable, TraversableResources
|
||||
|
||||
|
||||
class SimpleReader(abc.ABC):
|
||||
"""
|
||||
The minimum, low-level interface required from a resource
|
||||
provider.
|
||||
"""
|
||||
|
||||
@abc.abstractproperty
|
||||
def package(self):
|
||||
# type: () -> str
|
||||
"""
|
||||
The name of the package for which this reader loads resources.
|
||||
"""
|
||||
|
||||
@abc.abstractmethod
|
||||
def children(self):
|
||||
# type: () -> List['SimpleReader']
|
||||
"""
|
||||
Obtain an iterable of SimpleReader for available
|
||||
child containers (e.g. directories).
|
||||
"""
|
||||
|
||||
@abc.abstractmethod
|
||||
def resources(self):
|
||||
# type: () -> List[str]
|
||||
"""
|
||||
Obtain available named resources for this virtual package.
|
||||
"""
|
||||
|
||||
@abc.abstractmethod
|
||||
def open_binary(self, resource):
|
||||
# type: (str) -> BinaryIO
|
||||
"""
|
||||
Obtain a File-like for a named resource.
|
||||
"""
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
return self.package.split('.')[-1]
|
||||
|
||||
|
||||
class ResourceHandle(Traversable):
|
||||
"""
|
||||
Handle to a named resource in a ResourceReader.
|
||||
"""
|
||||
|
||||
def __init__(self, parent, name):
|
||||
# type: (ResourceContainer, str) -> None
|
||||
self.parent = parent
|
||||
self.name = name # type: ignore
|
||||
|
||||
def is_file(self):
|
||||
return True
|
||||
|
||||
def is_dir(self):
|
||||
return False
|
||||
|
||||
def open(self, mode='r', *args, **kwargs):
|
||||
stream = self.parent.reader.open_binary(self.name)
|
||||
if 'b' not in mode:
|
||||
stream = io.TextIOWrapper(*args, **kwargs)
|
||||
return stream
|
||||
|
||||
def joinpath(self, name):
|
||||
raise RuntimeError("Cannot traverse into a resource")
|
||||
|
||||
|
||||
class ResourceContainer(Traversable):
|
||||
"""
|
||||
Traversable container for a package's resources via its reader.
|
||||
"""
|
||||
|
||||
def __init__(self, reader):
|
||||
# type: (SimpleReader) -> None
|
||||
self.reader = reader
|
||||
|
||||
def is_dir(self):
|
||||
return True
|
||||
|
||||
def is_file(self):
|
||||
return False
|
||||
|
||||
def iterdir(self):
|
||||
files = (ResourceHandle(self, name) for name in self.reader.resources)
|
||||
dirs = map(ResourceContainer, self.reader.children())
|
||||
return itertools.chain(files, dirs)
|
||||
|
||||
def open(self, *args, **kwargs):
|
||||
raise IsADirectoryError()
|
||||
|
||||
def joinpath(self, name):
|
||||
return next(
|
||||
traversable for traversable in self.iterdir() if traversable.name == name
|
||||
)
|
||||
|
||||
|
||||
class TraversableReader(TraversableResources, SimpleReader):
|
||||
"""
|
||||
A TraversableResources based on SimpleReader. Resource providers
|
||||
may derive from this class to provide the TraversableResources
|
||||
interface by supplying the SimpleReader interface.
|
||||
"""
|
||||
|
||||
def files(self):
|
||||
return ResourceContainer(self)
|
||||
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
@@ -1,253 +0,0 @@
|
||||
import os
|
||||
import subprocess
|
||||
import contextlib
|
||||
import functools
|
||||
import tempfile
|
||||
import shutil
|
||||
import operator
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def pushd(dir):
|
||||
orig = os.getcwd()
|
||||
os.chdir(dir)
|
||||
try:
|
||||
yield dir
|
||||
finally:
|
||||
os.chdir(orig)
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def tarball_context(url, target_dir=None, runner=None, pushd=pushd):
|
||||
"""
|
||||
Get a tarball, extract it, change to that directory, yield, then
|
||||
clean up.
|
||||
`runner` is the function to invoke commands.
|
||||
`pushd` is a context manager for changing the directory.
|
||||
"""
|
||||
if target_dir is None:
|
||||
target_dir = os.path.basename(url).replace('.tar.gz', '').replace('.tgz', '')
|
||||
if runner is None:
|
||||
runner = functools.partial(subprocess.check_call, shell=True)
|
||||
# In the tar command, use --strip-components=1 to strip the first path and
|
||||
# then
|
||||
# use -C to cause the files to be extracted to {target_dir}. This ensures
|
||||
# that we always know where the files were extracted.
|
||||
runner('mkdir {target_dir}'.format(**vars()))
|
||||
try:
|
||||
getter = 'wget {url} -O -'
|
||||
extract = 'tar x{compression} --strip-components=1 -C {target_dir}'
|
||||
cmd = ' | '.join((getter, extract))
|
||||
runner(cmd.format(compression=infer_compression(url), **vars()))
|
||||
with pushd(target_dir):
|
||||
yield target_dir
|
||||
finally:
|
||||
runner('rm -Rf {target_dir}'.format(**vars()))
|
||||
|
||||
|
||||
def infer_compression(url):
|
||||
"""
|
||||
Given a URL or filename, infer the compression code for tar.
|
||||
"""
|
||||
# cheat and just assume it's the last two characters
|
||||
compression_indicator = url[-2:]
|
||||
mapping = dict(gz='z', bz='j', xz='J')
|
||||
# Assume 'z' (gzip) if no match
|
||||
return mapping.get(compression_indicator, 'z')
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def temp_dir(remover=shutil.rmtree):
|
||||
"""
|
||||
Create a temporary directory context. Pass a custom remover
|
||||
to override the removal behavior.
|
||||
"""
|
||||
temp_dir = tempfile.mkdtemp()
|
||||
try:
|
||||
yield temp_dir
|
||||
finally:
|
||||
remover(temp_dir)
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def repo_context(url, branch=None, quiet=True, dest_ctx=temp_dir):
|
||||
"""
|
||||
Check out the repo indicated by url.
|
||||
|
||||
If dest_ctx is supplied, it should be a context manager
|
||||
to yield the target directory for the check out.
|
||||
"""
|
||||
exe = 'git' if 'git' in url else 'hg'
|
||||
with dest_ctx() as repo_dir:
|
||||
cmd = [exe, 'clone', url, repo_dir]
|
||||
if branch:
|
||||
cmd.extend(['--branch', branch])
|
||||
devnull = open(os.path.devnull, 'w')
|
||||
stdout = devnull if quiet else None
|
||||
subprocess.check_call(cmd, stdout=stdout)
|
||||
yield repo_dir
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def null():
|
||||
yield
|
||||
|
||||
|
||||
class ExceptionTrap:
|
||||
"""
|
||||
A context manager that will catch certain exceptions and provide an
|
||||
indication they occurred.
|
||||
|
||||
>>> with ExceptionTrap() as trap:
|
||||
... raise Exception()
|
||||
>>> bool(trap)
|
||||
True
|
||||
|
||||
>>> with ExceptionTrap() as trap:
|
||||
... pass
|
||||
>>> bool(trap)
|
||||
False
|
||||
|
||||
>>> with ExceptionTrap(ValueError) as trap:
|
||||
... raise ValueError("1 + 1 is not 3")
|
||||
>>> bool(trap)
|
||||
True
|
||||
|
||||
>>> with ExceptionTrap(ValueError) as trap:
|
||||
... raise Exception()
|
||||
Traceback (most recent call last):
|
||||
...
|
||||
Exception
|
||||
|
||||
>>> bool(trap)
|
||||
False
|
||||
"""
|
||||
|
||||
exc_info = None, None, None
|
||||
|
||||
def __init__(self, exceptions=(Exception,)):
|
||||
self.exceptions = exceptions
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
@property
|
||||
def type(self):
|
||||
return self.exc_info[0]
|
||||
|
||||
@property
|
||||
def value(self):
|
||||
return self.exc_info[1]
|
||||
|
||||
@property
|
||||
def tb(self):
|
||||
return self.exc_info[2]
|
||||
|
||||
def __exit__(self, *exc_info):
|
||||
type = exc_info[0]
|
||||
matches = type and issubclass(type, self.exceptions)
|
||||
if matches:
|
||||
self.exc_info = exc_info
|
||||
return matches
|
||||
|
||||
def __bool__(self):
|
||||
return bool(self.type)
|
||||
|
||||
def raises(self, func, *, _test=bool):
|
||||
"""
|
||||
Wrap func and replace the result with the truth
|
||||
value of the trap (True if an exception occurred).
|
||||
|
||||
First, give the decorator an alias to support Python 3.8
|
||||
Syntax.
|
||||
|
||||
>>> raises = ExceptionTrap(ValueError).raises
|
||||
|
||||
Now decorate a function that always fails.
|
||||
|
||||
>>> @raises
|
||||
... def fail():
|
||||
... raise ValueError('failed')
|
||||
>>> fail()
|
||||
True
|
||||
"""
|
||||
|
||||
@functools.wraps(func)
|
||||
def wrapper(*args, **kwargs):
|
||||
with ExceptionTrap(self.exceptions) as trap:
|
||||
func(*args, **kwargs)
|
||||
return _test(trap)
|
||||
|
||||
return wrapper
|
||||
|
||||
def passes(self, func):
|
||||
"""
|
||||
Wrap func and replace the result with the truth
|
||||
value of the trap (True if no exception).
|
||||
|
||||
First, give the decorator an alias to support Python 3.8
|
||||
Syntax.
|
||||
|
||||
>>> passes = ExceptionTrap(ValueError).passes
|
||||
|
||||
Now decorate a function that always fails.
|
||||
|
||||
>>> @passes
|
||||
... def fail():
|
||||
... raise ValueError('failed')
|
||||
|
||||
>>> fail()
|
||||
False
|
||||
"""
|
||||
return self.raises(func, _test=operator.not_)
|
||||
|
||||
|
||||
class suppress(contextlib.suppress, contextlib.ContextDecorator):
|
||||
"""
|
||||
A version of contextlib.suppress with decorator support.
|
||||
|
||||
>>> @suppress(KeyError)
|
||||
... def key_error():
|
||||
... {}['']
|
||||
>>> key_error()
|
||||
"""
|
||||
|
||||
|
||||
class on_interrupt(contextlib.ContextDecorator):
|
||||
"""
|
||||
Replace a KeyboardInterrupt with SystemExit(1)
|
||||
|
||||
>>> def do_interrupt():
|
||||
... raise KeyboardInterrupt()
|
||||
>>> on_interrupt('error')(do_interrupt)()
|
||||
Traceback (most recent call last):
|
||||
...
|
||||
SystemExit: 1
|
||||
>>> on_interrupt('error', code=255)(do_interrupt)()
|
||||
Traceback (most recent call last):
|
||||
...
|
||||
SystemExit: 255
|
||||
>>> on_interrupt('suppress')(do_interrupt)()
|
||||
>>> with __import__('pytest').raises(KeyboardInterrupt):
|
||||
... on_interrupt('ignore')(do_interrupt)()
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
action='error',
|
||||
# py3.7 compat
|
||||
# /,
|
||||
code=1,
|
||||
):
|
||||
self.action = action
|
||||
self.code = code
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exctype, excinst, exctb):
|
||||
if exctype is not KeyboardInterrupt or self.action == 'ignore':
|
||||
return
|
||||
elif self.action == 'error':
|
||||
raise SystemExit(self.code) from excinst
|
||||
return self.action == 'suppress'
|
||||
@@ -1,525 +0,0 @@
|
||||
import functools
|
||||
import time
|
||||
import inspect
|
||||
import collections
|
||||
import types
|
||||
import itertools
|
||||
|
||||
import setuptools.extern.more_itertools
|
||||
|
||||
from typing import Callable, TypeVar
|
||||
|
||||
|
||||
CallableT = TypeVar("CallableT", bound=Callable[..., object])
|
||||
|
||||
|
||||
def compose(*funcs):
|
||||
"""
|
||||
Compose any number of unary functions into a single unary function.
|
||||
|
||||
>>> import textwrap
|
||||
>>> expected = str.strip(textwrap.dedent(compose.__doc__))
|
||||
>>> strip_and_dedent = compose(str.strip, textwrap.dedent)
|
||||
>>> strip_and_dedent(compose.__doc__) == expected
|
||||
True
|
||||
|
||||
Compose also allows the innermost function to take arbitrary arguments.
|
||||
|
||||
>>> round_three = lambda x: round(x, ndigits=3)
|
||||
>>> f = compose(round_three, int.__truediv__)
|
||||
>>> [f(3*x, x+1) for x in range(1,10)]
|
||||
[1.5, 2.0, 2.25, 2.4, 2.5, 2.571, 2.625, 2.667, 2.7]
|
||||
"""
|
||||
|
||||
def compose_two(f1, f2):
|
||||
return lambda *args, **kwargs: f1(f2(*args, **kwargs))
|
||||
|
||||
return functools.reduce(compose_two, funcs)
|
||||
|
||||
|
||||
def method_caller(method_name, *args, **kwargs):
|
||||
"""
|
||||
Return a function that will call a named method on the
|
||||
target object with optional positional and keyword
|
||||
arguments.
|
||||
|
||||
>>> lower = method_caller('lower')
|
||||
>>> lower('MyString')
|
||||
'mystring'
|
||||
"""
|
||||
|
||||
def call_method(target):
|
||||
func = getattr(target, method_name)
|
||||
return func(*args, **kwargs)
|
||||
|
||||
return call_method
|
||||
|
||||
|
||||
def once(func):
|
||||
"""
|
||||
Decorate func so it's only ever called the first time.
|
||||
|
||||
This decorator can ensure that an expensive or non-idempotent function
|
||||
will not be expensive on subsequent calls and is idempotent.
|
||||
|
||||
>>> add_three = once(lambda a: a+3)
|
||||
>>> add_three(3)
|
||||
6
|
||||
>>> add_three(9)
|
||||
6
|
||||
>>> add_three('12')
|
||||
6
|
||||
|
||||
To reset the stored value, simply clear the property ``saved_result``.
|
||||
|
||||
>>> del add_three.saved_result
|
||||
>>> add_three(9)
|
||||
12
|
||||
>>> add_three(8)
|
||||
12
|
||||
|
||||
Or invoke 'reset()' on it.
|
||||
|
||||
>>> add_three.reset()
|
||||
>>> add_three(-3)
|
||||
0
|
||||
>>> add_three(0)
|
||||
0
|
||||
"""
|
||||
|
||||
@functools.wraps(func)
|
||||
def wrapper(*args, **kwargs):
|
||||
if not hasattr(wrapper, 'saved_result'):
|
||||
wrapper.saved_result = func(*args, **kwargs)
|
||||
return wrapper.saved_result
|
||||
|
||||
wrapper.reset = lambda: vars(wrapper).__delitem__('saved_result')
|
||||
return wrapper
|
||||
|
||||
|
||||
def method_cache(
|
||||
method: CallableT,
|
||||
cache_wrapper: Callable[
|
||||
[CallableT], CallableT
|
||||
] = functools.lru_cache(), # type: ignore[assignment]
|
||||
) -> CallableT:
|
||||
"""
|
||||
Wrap lru_cache to support storing the cache data in the object instances.
|
||||
|
||||
Abstracts the common paradigm where the method explicitly saves an
|
||||
underscore-prefixed protected property on first call and returns that
|
||||
subsequently.
|
||||
|
||||
>>> class MyClass:
|
||||
... calls = 0
|
||||
...
|
||||
... @method_cache
|
||||
... def method(self, value):
|
||||
... self.calls += 1
|
||||
... return value
|
||||
|
||||
>>> a = MyClass()
|
||||
>>> a.method(3)
|
||||
3
|
||||
>>> for x in range(75):
|
||||
... res = a.method(x)
|
||||
>>> a.calls
|
||||
75
|
||||
|
||||
Note that the apparent behavior will be exactly like that of lru_cache
|
||||
except that the cache is stored on each instance, so values in one
|
||||
instance will not flush values from another, and when an instance is
|
||||
deleted, so are the cached values for that instance.
|
||||
|
||||
>>> b = MyClass()
|
||||
>>> for x in range(35):
|
||||
... res = b.method(x)
|
||||
>>> b.calls
|
||||
35
|
||||
>>> a.method(0)
|
||||
0
|
||||
>>> a.calls
|
||||
75
|
||||
|
||||
Note that if method had been decorated with ``functools.lru_cache()``,
|
||||
a.calls would have been 76 (due to the cached value of 0 having been
|
||||
flushed by the 'b' instance).
|
||||
|
||||
Clear the cache with ``.cache_clear()``
|
||||
|
||||
>>> a.method.cache_clear()
|
||||
|
||||
Same for a method that hasn't yet been called.
|
||||
|
||||
>>> c = MyClass()
|
||||
>>> c.method.cache_clear()
|
||||
|
||||
Another cache wrapper may be supplied:
|
||||
|
||||
>>> cache = functools.lru_cache(maxsize=2)
|
||||
>>> MyClass.method2 = method_cache(lambda self: 3, cache_wrapper=cache)
|
||||
>>> a = MyClass()
|
||||
>>> a.method2()
|
||||
3
|
||||
|
||||
Caution - do not subsequently wrap the method with another decorator, such
|
||||
as ``@property``, which changes the semantics of the function.
|
||||
|
||||
See also
|
||||
http://code.activestate.com/recipes/577452-a-memoize-decorator-for-instance-methods/
|
||||
for another implementation and additional justification.
|
||||
"""
|
||||
|
||||
def wrapper(self: object, *args: object, **kwargs: object) -> object:
|
||||
# it's the first call, replace the method with a cached, bound method
|
||||
bound_method: CallableT = types.MethodType( # type: ignore[assignment]
|
||||
method, self
|
||||
)
|
||||
cached_method = cache_wrapper(bound_method)
|
||||
setattr(self, method.__name__, cached_method)
|
||||
return cached_method(*args, **kwargs)
|
||||
|
||||
# Support cache clear even before cache has been created.
|
||||
wrapper.cache_clear = lambda: None # type: ignore[attr-defined]
|
||||
|
||||
return ( # type: ignore[return-value]
|
||||
_special_method_cache(method, cache_wrapper) or wrapper
|
||||
)
|
||||
|
||||
|
||||
def _special_method_cache(method, cache_wrapper):
|
||||
"""
|
||||
Because Python treats special methods differently, it's not
|
||||
possible to use instance attributes to implement the cached
|
||||
methods.
|
||||
|
||||
Instead, install the wrapper method under a different name
|
||||
and return a simple proxy to that wrapper.
|
||||
|
||||
https://github.com/jaraco/jaraco.functools/issues/5
|
||||
"""
|
||||
name = method.__name__
|
||||
special_names = '__getattr__', '__getitem__'
|
||||
if name not in special_names:
|
||||
return
|
||||
|
||||
wrapper_name = '__cached' + name
|
||||
|
||||
def proxy(self, *args, **kwargs):
|
||||
if wrapper_name not in vars(self):
|
||||
bound = types.MethodType(method, self)
|
||||
cache = cache_wrapper(bound)
|
||||
setattr(self, wrapper_name, cache)
|
||||
else:
|
||||
cache = getattr(self, wrapper_name)
|
||||
return cache(*args, **kwargs)
|
||||
|
||||
return proxy
|
||||
|
||||
|
||||
def apply(transform):
|
||||
"""
|
||||
Decorate a function with a transform function that is
|
||||
invoked on results returned from the decorated function.
|
||||
|
||||
>>> @apply(reversed)
|
||||
... def get_numbers(start):
|
||||
... "doc for get_numbers"
|
||||
... return range(start, start+3)
|
||||
>>> list(get_numbers(4))
|
||||
[6, 5, 4]
|
||||
>>> get_numbers.__doc__
|
||||
'doc for get_numbers'
|
||||
"""
|
||||
|
||||
def wrap(func):
|
||||
return functools.wraps(func)(compose(transform, func))
|
||||
|
||||
return wrap
|
||||
|
||||
|
||||
def result_invoke(action):
|
||||
r"""
|
||||
Decorate a function with an action function that is
|
||||
invoked on the results returned from the decorated
|
||||
function (for its side-effect), then return the original
|
||||
result.
|
||||
|
||||
>>> @result_invoke(print)
|
||||
... def add_two(a, b):
|
||||
... return a + b
|
||||
>>> x = add_two(2, 3)
|
||||
5
|
||||
>>> x
|
||||
5
|
||||
"""
|
||||
|
||||
def wrap(func):
|
||||
@functools.wraps(func)
|
||||
def wrapper(*args, **kwargs):
|
||||
result = func(*args, **kwargs)
|
||||
action(result)
|
||||
return result
|
||||
|
||||
return wrapper
|
||||
|
||||
return wrap
|
||||
|
||||
|
||||
def call_aside(f, *args, **kwargs):
|
||||
"""
|
||||
Call a function for its side effect after initialization.
|
||||
|
||||
>>> @call_aside
|
||||
... def func(): print("called")
|
||||
called
|
||||
>>> func()
|
||||
called
|
||||
|
||||
Use functools.partial to pass parameters to the initial call
|
||||
|
||||
>>> @functools.partial(call_aside, name='bingo')
|
||||
... def func(name): print("called with", name)
|
||||
called with bingo
|
||||
"""
|
||||
f(*args, **kwargs)
|
||||
return f
|
||||
|
||||
|
||||
class Throttler:
|
||||
"""
|
||||
Rate-limit a function (or other callable)
|
||||
"""
|
||||
|
||||
def __init__(self, func, max_rate=float('Inf')):
|
||||
if isinstance(func, Throttler):
|
||||
func = func.func
|
||||
self.func = func
|
||||
self.max_rate = max_rate
|
||||
self.reset()
|
||||
|
||||
def reset(self):
|
||||
self.last_called = 0
|
||||
|
||||
def __call__(self, *args, **kwargs):
|
||||
self._wait()
|
||||
return self.func(*args, **kwargs)
|
||||
|
||||
def _wait(self):
|
||||
"ensure at least 1/max_rate seconds from last call"
|
||||
elapsed = time.time() - self.last_called
|
||||
must_wait = 1 / self.max_rate - elapsed
|
||||
time.sleep(max(0, must_wait))
|
||||
self.last_called = time.time()
|
||||
|
||||
def __get__(self, obj, type=None):
|
||||
return first_invoke(self._wait, functools.partial(self.func, obj))
|
||||
|
||||
|
||||
def first_invoke(func1, func2):
|
||||
"""
|
||||
Return a function that when invoked will invoke func1 without
|
||||
any parameters (for its side-effect) and then invoke func2
|
||||
with whatever parameters were passed, returning its result.
|
||||
"""
|
||||
|
||||
def wrapper(*args, **kwargs):
|
||||
func1()
|
||||
return func2(*args, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
def retry_call(func, cleanup=lambda: None, retries=0, trap=()):
|
||||
"""
|
||||
Given a callable func, trap the indicated exceptions
|
||||
for up to 'retries' times, invoking cleanup on the
|
||||
exception. On the final attempt, allow any exceptions
|
||||
to propagate.
|
||||
"""
|
||||
attempts = itertools.count() if retries == float('inf') else range(retries)
|
||||
for attempt in attempts:
|
||||
try:
|
||||
return func()
|
||||
except trap:
|
||||
cleanup()
|
||||
|
||||
return func()
|
||||
|
||||
|
||||
def retry(*r_args, **r_kwargs):
|
||||
"""
|
||||
Decorator wrapper for retry_call. Accepts arguments to retry_call
|
||||
except func and then returns a decorator for the decorated function.
|
||||
|
||||
Ex:
|
||||
|
||||
>>> @retry(retries=3)
|
||||
... def my_func(a, b):
|
||||
... "this is my funk"
|
||||
... print(a, b)
|
||||
>>> my_func.__doc__
|
||||
'this is my funk'
|
||||
"""
|
||||
|
||||
def decorate(func):
|
||||
@functools.wraps(func)
|
||||
def wrapper(*f_args, **f_kwargs):
|
||||
bound = functools.partial(func, *f_args, **f_kwargs)
|
||||
return retry_call(bound, *r_args, **r_kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
return decorate
|
||||
|
||||
|
||||
def print_yielded(func):
|
||||
"""
|
||||
Convert a generator into a function that prints all yielded elements
|
||||
|
||||
>>> @print_yielded
|
||||
... def x():
|
||||
... yield 3; yield None
|
||||
>>> x()
|
||||
3
|
||||
None
|
||||
"""
|
||||
print_all = functools.partial(map, print)
|
||||
print_results = compose(more_itertools.consume, print_all, func)
|
||||
return functools.wraps(func)(print_results)
|
||||
|
||||
|
||||
def pass_none(func):
|
||||
"""
|
||||
Wrap func so it's not called if its first param is None
|
||||
|
||||
>>> print_text = pass_none(print)
|
||||
>>> print_text('text')
|
||||
text
|
||||
>>> print_text(None)
|
||||
"""
|
||||
|
||||
@functools.wraps(func)
|
||||
def wrapper(param, *args, **kwargs):
|
||||
if param is not None:
|
||||
return func(param, *args, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
def assign_params(func, namespace):
|
||||
"""
|
||||
Assign parameters from namespace where func solicits.
|
||||
|
||||
>>> def func(x, y=3):
|
||||
... print(x, y)
|
||||
>>> assigned = assign_params(func, dict(x=2, z=4))
|
||||
>>> assigned()
|
||||
2 3
|
||||
|
||||
The usual errors are raised if a function doesn't receive
|
||||
its required parameters:
|
||||
|
||||
>>> assigned = assign_params(func, dict(y=3, z=4))
|
||||
>>> assigned()
|
||||
Traceback (most recent call last):
|
||||
TypeError: func() ...argument...
|
||||
|
||||
It even works on methods:
|
||||
|
||||
>>> class Handler:
|
||||
... def meth(self, arg):
|
||||
... print(arg)
|
||||
>>> assign_params(Handler().meth, dict(arg='crystal', foo='clear'))()
|
||||
crystal
|
||||
"""
|
||||
sig = inspect.signature(func)
|
||||
params = sig.parameters.keys()
|
||||
call_ns = {k: namespace[k] for k in params if k in namespace}
|
||||
return functools.partial(func, **call_ns)
|
||||
|
||||
|
||||
def save_method_args(method):
|
||||
"""
|
||||
Wrap a method such that when it is called, the args and kwargs are
|
||||
saved on the method.
|
||||
|
||||
>>> class MyClass:
|
||||
... @save_method_args
|
||||
... def method(self, a, b):
|
||||
... print(a, b)
|
||||
>>> my_ob = MyClass()
|
||||
>>> my_ob.method(1, 2)
|
||||
1 2
|
||||
>>> my_ob._saved_method.args
|
||||
(1, 2)
|
||||
>>> my_ob._saved_method.kwargs
|
||||
{}
|
||||
>>> my_ob.method(a=3, b='foo')
|
||||
3 foo
|
||||
>>> my_ob._saved_method.args
|
||||
()
|
||||
>>> my_ob._saved_method.kwargs == dict(a=3, b='foo')
|
||||
True
|
||||
|
||||
The arguments are stored on the instance, allowing for
|
||||
different instance to save different args.
|
||||
|
||||
>>> your_ob = MyClass()
|
||||
>>> your_ob.method({str('x'): 3}, b=[4])
|
||||
{'x': 3} [4]
|
||||
>>> your_ob._saved_method.args
|
||||
({'x': 3},)
|
||||
>>> my_ob._saved_method.args
|
||||
()
|
||||
"""
|
||||
args_and_kwargs = collections.namedtuple('args_and_kwargs', 'args kwargs')
|
||||
|
||||
@functools.wraps(method)
|
||||
def wrapper(self, *args, **kwargs):
|
||||
attr_name = '_saved_' + method.__name__
|
||||
attr = args_and_kwargs(args, kwargs)
|
||||
setattr(self, attr_name, attr)
|
||||
return method(self, *args, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
def except_(*exceptions, replace=None, use=None):
|
||||
"""
|
||||
Replace the indicated exceptions, if raised, with the indicated
|
||||
literal replacement or evaluated expression (if present).
|
||||
|
||||
>>> safe_int = except_(ValueError)(int)
|
||||
>>> safe_int('five')
|
||||
>>> safe_int('5')
|
||||
5
|
||||
|
||||
Specify a literal replacement with ``replace``.
|
||||
|
||||
>>> safe_int_r = except_(ValueError, replace=0)(int)
|
||||
>>> safe_int_r('five')
|
||||
0
|
||||
|
||||
Provide an expression to ``use`` to pass through particular parameters.
|
||||
|
||||
>>> safe_int_pt = except_(ValueError, use='args[0]')(int)
|
||||
>>> safe_int_pt('five')
|
||||
'five'
|
||||
|
||||
"""
|
||||
|
||||
def decorate(func):
|
||||
@functools.wraps(func)
|
||||
def wrapper(*args, **kwargs):
|
||||
try:
|
||||
return func(*args, **kwargs)
|
||||
except exceptions:
|
||||
try:
|
||||
return eval(use)
|
||||
except TypeError:
|
||||
return replace
|
||||
|
||||
return wrapper
|
||||
|
||||
return decorate
|
||||
@@ -1,599 +0,0 @@
|
||||
import re
|
||||
import itertools
|
||||
import textwrap
|
||||
import functools
|
||||
|
||||
try:
|
||||
from importlib.resources import files # type: ignore
|
||||
except ImportError: # pragma: nocover
|
||||
from setuptools.extern.importlib_resources import files # type: ignore
|
||||
|
||||
from setuptools.extern.jaraco.functools import compose, method_cache
|
||||
from setuptools.extern.jaraco.context import ExceptionTrap
|
||||
|
||||
|
||||
def substitution(old, new):
|
||||
"""
|
||||
Return a function that will perform a substitution on a string
|
||||
"""
|
||||
return lambda s: s.replace(old, new)
|
||||
|
||||
|
||||
def multi_substitution(*substitutions):
|
||||
"""
|
||||
Take a sequence of pairs specifying substitutions, and create
|
||||
a function that performs those substitutions.
|
||||
|
||||
>>> multi_substitution(('foo', 'bar'), ('bar', 'baz'))('foo')
|
||||
'baz'
|
||||
"""
|
||||
substitutions = itertools.starmap(substitution, substitutions)
|
||||
# compose function applies last function first, so reverse the
|
||||
# substitutions to get the expected order.
|
||||
substitutions = reversed(tuple(substitutions))
|
||||
return compose(*substitutions)
|
||||
|
||||
|
||||
class FoldedCase(str):
|
||||
"""
|
||||
A case insensitive string class; behaves just like str
|
||||
except compares equal when the only variation is case.
|
||||
|
||||
>>> s = FoldedCase('hello world')
|
||||
|
||||
>>> s == 'Hello World'
|
||||
True
|
||||
|
||||
>>> 'Hello World' == s
|
||||
True
|
||||
|
||||
>>> s != 'Hello World'
|
||||
False
|
||||
|
||||
>>> s.index('O')
|
||||
4
|
||||
|
||||
>>> s.split('O')
|
||||
['hell', ' w', 'rld']
|
||||
|
||||
>>> sorted(map(FoldedCase, ['GAMMA', 'alpha', 'Beta']))
|
||||
['alpha', 'Beta', 'GAMMA']
|
||||
|
||||
Sequence membership is straightforward.
|
||||
|
||||
>>> "Hello World" in [s]
|
||||
True
|
||||
>>> s in ["Hello World"]
|
||||
True
|
||||
|
||||
You may test for set inclusion, but candidate and elements
|
||||
must both be folded.
|
||||
|
||||
>>> FoldedCase("Hello World") in {s}
|
||||
True
|
||||
>>> s in {FoldedCase("Hello World")}
|
||||
True
|
||||
|
||||
String inclusion works as long as the FoldedCase object
|
||||
is on the right.
|
||||
|
||||
>>> "hello" in FoldedCase("Hello World")
|
||||
True
|
||||
|
||||
But not if the FoldedCase object is on the left:
|
||||
|
||||
>>> FoldedCase('hello') in 'Hello World'
|
||||
False
|
||||
|
||||
In that case, use ``in_``:
|
||||
|
||||
>>> FoldedCase('hello').in_('Hello World')
|
||||
True
|
||||
|
||||
>>> FoldedCase('hello') > FoldedCase('Hello')
|
||||
False
|
||||
"""
|
||||
|
||||
def __lt__(self, other):
|
||||
return self.lower() < other.lower()
|
||||
|
||||
def __gt__(self, other):
|
||||
return self.lower() > other.lower()
|
||||
|
||||
def __eq__(self, other):
|
||||
return self.lower() == other.lower()
|
||||
|
||||
def __ne__(self, other):
|
||||
return self.lower() != other.lower()
|
||||
|
||||
def __hash__(self):
|
||||
return hash(self.lower())
|
||||
|
||||
def __contains__(self, other):
|
||||
return super().lower().__contains__(other.lower())
|
||||
|
||||
def in_(self, other):
|
||||
"Does self appear in other?"
|
||||
return self in FoldedCase(other)
|
||||
|
||||
# cache lower since it's likely to be called frequently.
|
||||
@method_cache
|
||||
def lower(self):
|
||||
return super().lower()
|
||||
|
||||
def index(self, sub):
|
||||
return self.lower().index(sub.lower())
|
||||
|
||||
def split(self, splitter=' ', maxsplit=0):
|
||||
pattern = re.compile(re.escape(splitter), re.I)
|
||||
return pattern.split(self, maxsplit)
|
||||
|
||||
|
||||
# Python 3.8 compatibility
|
||||
_unicode_trap = ExceptionTrap(UnicodeDecodeError)
|
||||
|
||||
|
||||
@_unicode_trap.passes
|
||||
def is_decodable(value):
|
||||
r"""
|
||||
Return True if the supplied value is decodable (using the default
|
||||
encoding).
|
||||
|
||||
>>> is_decodable(b'\xff')
|
||||
False
|
||||
>>> is_decodable(b'\x32')
|
||||
True
|
||||
"""
|
||||
value.decode()
|
||||
|
||||
|
||||
def is_binary(value):
|
||||
r"""
|
||||
Return True if the value appears to be binary (that is, it's a byte
|
||||
string and isn't decodable).
|
||||
|
||||
>>> is_binary(b'\xff')
|
||||
True
|
||||
>>> is_binary('\xff')
|
||||
False
|
||||
"""
|
||||
return isinstance(value, bytes) and not is_decodable(value)
|
||||
|
||||
|
||||
def trim(s):
|
||||
r"""
|
||||
Trim something like a docstring to remove the whitespace that
|
||||
is common due to indentation and formatting.
|
||||
|
||||
>>> trim("\n\tfoo = bar\n\t\tbar = baz\n")
|
||||
'foo = bar\n\tbar = baz'
|
||||
"""
|
||||
return textwrap.dedent(s).strip()
|
||||
|
||||
|
||||
def wrap(s):
|
||||
"""
|
||||
Wrap lines of text, retaining existing newlines as
|
||||
paragraph markers.
|
||||
|
||||
>>> print(wrap(lorem_ipsum))
|
||||
Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do
|
||||
eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad
|
||||
minim veniam, quis nostrud exercitation ullamco laboris nisi ut
|
||||
aliquip ex ea commodo consequat. Duis aute irure dolor in
|
||||
reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla
|
||||
pariatur. Excepteur sint occaecat cupidatat non proident, sunt in
|
||||
culpa qui officia deserunt mollit anim id est laborum.
|
||||
<BLANKLINE>
|
||||
Curabitur pretium tincidunt lacus. Nulla gravida orci a odio. Nullam
|
||||
varius, turpis et commodo pharetra, est eros bibendum elit, nec luctus
|
||||
magna felis sollicitudin mauris. Integer in mauris eu nibh euismod
|
||||
gravida. Duis ac tellus et risus vulputate vehicula. Donec lobortis
|
||||
risus a elit. Etiam tempor. Ut ullamcorper, ligula eu tempor congue,
|
||||
eros est euismod turpis, id tincidunt sapien risus a quam. Maecenas
|
||||
fermentum consequat mi. Donec fermentum. Pellentesque malesuada nulla
|
||||
a mi. Duis sapien sem, aliquet nec, commodo eget, consequat quis,
|
||||
neque. Aliquam faucibus, elit ut dictum aliquet, felis nisl adipiscing
|
||||
sapien, sed malesuada diam lacus eget erat. Cras mollis scelerisque
|
||||
nunc. Nullam arcu. Aliquam consequat. Curabitur augue lorem, dapibus
|
||||
quis, laoreet et, pretium ac, nisi. Aenean magna nisl, mollis quis,
|
||||
molestie eu, feugiat in, orci. In hac habitasse platea dictumst.
|
||||
"""
|
||||
paragraphs = s.splitlines()
|
||||
wrapped = ('\n'.join(textwrap.wrap(para)) for para in paragraphs)
|
||||
return '\n\n'.join(wrapped)
|
||||
|
||||
|
||||
def unwrap(s):
|
||||
r"""
|
||||
Given a multi-line string, return an unwrapped version.
|
||||
|
||||
>>> wrapped = wrap(lorem_ipsum)
|
||||
>>> wrapped.count('\n')
|
||||
20
|
||||
>>> unwrapped = unwrap(wrapped)
|
||||
>>> unwrapped.count('\n')
|
||||
1
|
||||
>>> print(unwrapped)
|
||||
Lorem ipsum dolor sit amet, consectetur adipiscing ...
|
||||
Curabitur pretium tincidunt lacus. Nulla gravida orci ...
|
||||
|
||||
"""
|
||||
paragraphs = re.split(r'\n\n+', s)
|
||||
cleaned = (para.replace('\n', ' ') for para in paragraphs)
|
||||
return '\n'.join(cleaned)
|
||||
|
||||
|
||||
|
||||
|
||||
class Splitter(object):
|
||||
"""object that will split a string with the given arguments for each call
|
||||
|
||||
>>> s = Splitter(',')
|
||||
>>> s('hello, world, this is your, master calling')
|
||||
['hello', ' world', ' this is your', ' master calling']
|
||||
"""
|
||||
|
||||
def __init__(self, *args):
|
||||
self.args = args
|
||||
|
||||
def __call__(self, s):
|
||||
return s.split(*self.args)
|
||||
|
||||
|
||||
def indent(string, prefix=' ' * 4):
|
||||
"""
|
||||
>>> indent('foo')
|
||||
' foo'
|
||||
"""
|
||||
return prefix + string
|
||||
|
||||
|
||||
class WordSet(tuple):
|
||||
"""
|
||||
Given an identifier, return the words that identifier represents,
|
||||
whether in camel case, underscore-separated, etc.
|
||||
|
||||
>>> WordSet.parse("camelCase")
|
||||
('camel', 'Case')
|
||||
|
||||
>>> WordSet.parse("under_sep")
|
||||
('under', 'sep')
|
||||
|
||||
Acronyms should be retained
|
||||
|
||||
>>> WordSet.parse("firstSNL")
|
||||
('first', 'SNL')
|
||||
|
||||
>>> WordSet.parse("you_and_I")
|
||||
('you', 'and', 'I')
|
||||
|
||||
>>> WordSet.parse("A simple test")
|
||||
('A', 'simple', 'test')
|
||||
|
||||
Multiple caps should not interfere with the first cap of another word.
|
||||
|
||||
>>> WordSet.parse("myABCClass")
|
||||
('my', 'ABC', 'Class')
|
||||
|
||||
The result is a WordSet, so you can get the form you need.
|
||||
|
||||
>>> WordSet.parse("myABCClass").underscore_separated()
|
||||
'my_ABC_Class'
|
||||
|
||||
>>> WordSet.parse('a-command').camel_case()
|
||||
'ACommand'
|
||||
|
||||
>>> WordSet.parse('someIdentifier').lowered().space_separated()
|
||||
'some identifier'
|
||||
|
||||
Slices of the result should return another WordSet.
|
||||
|
||||
>>> WordSet.parse('taken-out-of-context')[1:].underscore_separated()
|
||||
'out_of_context'
|
||||
|
||||
>>> WordSet.from_class_name(WordSet()).lowered().space_separated()
|
||||
'word set'
|
||||
|
||||
>>> example = WordSet.parse('figured it out')
|
||||
>>> example.headless_camel_case()
|
||||
'figuredItOut'
|
||||
>>> example.dash_separated()
|
||||
'figured-it-out'
|
||||
|
||||
"""
|
||||
|
||||
_pattern = re.compile('([A-Z]?[a-z]+)|([A-Z]+(?![a-z]))')
|
||||
|
||||
def capitalized(self):
|
||||
return WordSet(word.capitalize() for word in self)
|
||||
|
||||
def lowered(self):
|
||||
return WordSet(word.lower() for word in self)
|
||||
|
||||
def camel_case(self):
|
||||
return ''.join(self.capitalized())
|
||||
|
||||
def headless_camel_case(self):
|
||||
words = iter(self)
|
||||
first = next(words).lower()
|
||||
new_words = itertools.chain((first,), WordSet(words).camel_case())
|
||||
return ''.join(new_words)
|
||||
|
||||
def underscore_separated(self):
|
||||
return '_'.join(self)
|
||||
|
||||
def dash_separated(self):
|
||||
return '-'.join(self)
|
||||
|
||||
def space_separated(self):
|
||||
return ' '.join(self)
|
||||
|
||||
def trim_right(self, item):
|
||||
"""
|
||||
Remove the item from the end of the set.
|
||||
|
||||
>>> WordSet.parse('foo bar').trim_right('foo')
|
||||
('foo', 'bar')
|
||||
>>> WordSet.parse('foo bar').trim_right('bar')
|
||||
('foo',)
|
||||
>>> WordSet.parse('').trim_right('bar')
|
||||
()
|
||||
"""
|
||||
return self[:-1] if self and self[-1] == item else self
|
||||
|
||||
def trim_left(self, item):
|
||||
"""
|
||||
Remove the item from the beginning of the set.
|
||||
|
||||
>>> WordSet.parse('foo bar').trim_left('foo')
|
||||
('bar',)
|
||||
>>> WordSet.parse('foo bar').trim_left('bar')
|
||||
('foo', 'bar')
|
||||
>>> WordSet.parse('').trim_left('bar')
|
||||
()
|
||||
"""
|
||||
return self[1:] if self and self[0] == item else self
|
||||
|
||||
def trim(self, item):
|
||||
"""
|
||||
>>> WordSet.parse('foo bar').trim('foo')
|
||||
('bar',)
|
||||
"""
|
||||
return self.trim_left(item).trim_right(item)
|
||||
|
||||
def __getitem__(self, item):
|
||||
result = super(WordSet, self).__getitem__(item)
|
||||
if isinstance(item, slice):
|
||||
result = WordSet(result)
|
||||
return result
|
||||
|
||||
@classmethod
|
||||
def parse(cls, identifier):
|
||||
matches = cls._pattern.finditer(identifier)
|
||||
return WordSet(match.group(0) for match in matches)
|
||||
|
||||
@classmethod
|
||||
def from_class_name(cls, subject):
|
||||
return cls.parse(subject.__class__.__name__)
|
||||
|
||||
|
||||
# for backward compatibility
|
||||
words = WordSet.parse
|
||||
|
||||
|
||||
def simple_html_strip(s):
|
||||
r"""
|
||||
Remove HTML from the string `s`.
|
||||
|
||||
>>> str(simple_html_strip(''))
|
||||
''
|
||||
|
||||
>>> print(simple_html_strip('A <bold>stormy</bold> day in paradise'))
|
||||
A stormy day in paradise
|
||||
|
||||
>>> print(simple_html_strip('Somebody <!-- do not --> tell the truth.'))
|
||||
Somebody tell the truth.
|
||||
|
||||
>>> print(simple_html_strip('What about<br/>\nmultiple lines?'))
|
||||
What about
|
||||
multiple lines?
|
||||
"""
|
||||
html_stripper = re.compile('(<!--.*?-->)|(<[^>]*>)|([^<]+)', re.DOTALL)
|
||||
texts = (match.group(3) or '' for match in html_stripper.finditer(s))
|
||||
return ''.join(texts)
|
||||
|
||||
|
||||
class SeparatedValues(str):
|
||||
"""
|
||||
A string separated by a separator. Overrides __iter__ for getting
|
||||
the values.
|
||||
|
||||
>>> list(SeparatedValues('a,b,c'))
|
||||
['a', 'b', 'c']
|
||||
|
||||
Whitespace is stripped and empty values are discarded.
|
||||
|
||||
>>> list(SeparatedValues(' a, b , c, '))
|
||||
['a', 'b', 'c']
|
||||
"""
|
||||
|
||||
separator = ','
|
||||
|
||||
def __iter__(self):
|
||||
parts = self.split(self.separator)
|
||||
return filter(None, (part.strip() for part in parts))
|
||||
|
||||
|
||||
class Stripper:
|
||||
r"""
|
||||
Given a series of lines, find the common prefix and strip it from them.
|
||||
|
||||
>>> lines = [
|
||||
... 'abcdefg\n',
|
||||
... 'abc\n',
|
||||
... 'abcde\n',
|
||||
... ]
|
||||
>>> res = Stripper.strip_prefix(lines)
|
||||
>>> res.prefix
|
||||
'abc'
|
||||
>>> list(res.lines)
|
||||
['defg\n', '\n', 'de\n']
|
||||
|
||||
If no prefix is common, nothing should be stripped.
|
||||
|
||||
>>> lines = [
|
||||
... 'abcd\n',
|
||||
... '1234\n',
|
||||
... ]
|
||||
>>> res = Stripper.strip_prefix(lines)
|
||||
>>> res.prefix = ''
|
||||
>>> list(res.lines)
|
||||
['abcd\n', '1234\n']
|
||||
"""
|
||||
|
||||
def __init__(self, prefix, lines):
|
||||
self.prefix = prefix
|
||||
self.lines = map(self, lines)
|
||||
|
||||
@classmethod
|
||||
def strip_prefix(cls, lines):
|
||||
prefix_lines, lines = itertools.tee(lines)
|
||||
prefix = functools.reduce(cls.common_prefix, prefix_lines)
|
||||
return cls(prefix, lines)
|
||||
|
||||
def __call__(self, line):
|
||||
if not self.prefix:
|
||||
return line
|
||||
null, prefix, rest = line.partition(self.prefix)
|
||||
return rest
|
||||
|
||||
@staticmethod
|
||||
def common_prefix(s1, s2):
|
||||
"""
|
||||
Return the common prefix of two lines.
|
||||
"""
|
||||
index = min(len(s1), len(s2))
|
||||
while s1[:index] != s2[:index]:
|
||||
index -= 1
|
||||
return s1[:index]
|
||||
|
||||
|
||||
def remove_prefix(text, prefix):
|
||||
"""
|
||||
Remove the prefix from the text if it exists.
|
||||
|
||||
>>> remove_prefix('underwhelming performance', 'underwhelming ')
|
||||
'performance'
|
||||
|
||||
>>> remove_prefix('something special', 'sample')
|
||||
'something special'
|
||||
"""
|
||||
null, prefix, rest = text.rpartition(prefix)
|
||||
return rest
|
||||
|
||||
|
||||
def remove_suffix(text, suffix):
|
||||
"""
|
||||
Remove the suffix from the text if it exists.
|
||||
|
||||
>>> remove_suffix('name.git', '.git')
|
||||
'name'
|
||||
|
||||
>>> remove_suffix('something special', 'sample')
|
||||
'something special'
|
||||
"""
|
||||
rest, suffix, null = text.partition(suffix)
|
||||
return rest
|
||||
|
||||
|
||||
def normalize_newlines(text):
|
||||
r"""
|
||||
Replace alternate newlines with the canonical newline.
|
||||
|
||||
>>> normalize_newlines('Lorem Ipsum\u2029')
|
||||
'Lorem Ipsum\n'
|
||||
>>> normalize_newlines('Lorem Ipsum\r\n')
|
||||
'Lorem Ipsum\n'
|
||||
>>> normalize_newlines('Lorem Ipsum\x85')
|
||||
'Lorem Ipsum\n'
|
||||
"""
|
||||
newlines = ['\r\n', '\r', '\n', '\u0085', '\u2028', '\u2029']
|
||||
pattern = '|'.join(newlines)
|
||||
return re.sub(pattern, '\n', text)
|
||||
|
||||
|
||||
def _nonblank(str):
|
||||
return str and not str.startswith('#')
|
||||
|
||||
|
||||
@functools.singledispatch
|
||||
def yield_lines(iterable):
|
||||
r"""
|
||||
Yield valid lines of a string or iterable.
|
||||
|
||||
>>> list(yield_lines(''))
|
||||
[]
|
||||
>>> list(yield_lines(['foo', 'bar']))
|
||||
['foo', 'bar']
|
||||
>>> list(yield_lines('foo\nbar'))
|
||||
['foo', 'bar']
|
||||
>>> list(yield_lines('\nfoo\n#bar\nbaz #comment'))
|
||||
['foo', 'baz #comment']
|
||||
>>> list(yield_lines(['foo\nbar', 'baz', 'bing\n\n\n']))
|
||||
['foo', 'bar', 'baz', 'bing']
|
||||
"""
|
||||
return itertools.chain.from_iterable(map(yield_lines, iterable))
|
||||
|
||||
|
||||
@yield_lines.register(str)
|
||||
def _(text):
|
||||
return filter(_nonblank, map(str.strip, text.splitlines()))
|
||||
|
||||
|
||||
def drop_comment(line):
|
||||
"""
|
||||
Drop comments.
|
||||
|
||||
>>> drop_comment('foo # bar')
|
||||
'foo'
|
||||
|
||||
A hash without a space may be in a URL.
|
||||
|
||||
>>> drop_comment('http://example.com/foo#bar')
|
||||
'http://example.com/foo#bar'
|
||||
"""
|
||||
return line.partition(' #')[0]
|
||||
|
||||
|
||||
def join_continuation(lines):
|
||||
r"""
|
||||
Join lines continued by a trailing backslash.
|
||||
|
||||
>>> list(join_continuation(['foo \\', 'bar', 'baz']))
|
||||
['foobar', 'baz']
|
||||
>>> list(join_continuation(['foo \\', 'bar', 'baz']))
|
||||
['foobar', 'baz']
|
||||
>>> list(join_continuation(['foo \\', 'bar \\', 'baz']))
|
||||
['foobarbaz']
|
||||
|
||||
Not sure why, but...
|
||||
The character preceeding the backslash is also elided.
|
||||
|
||||
>>> list(join_continuation(['goo\\', 'dly']))
|
||||
['godly']
|
||||
|
||||
A terrible idea, but...
|
||||
If no line is available to continue, suppress the lines.
|
||||
|
||||
>>> list(join_continuation(['foo', 'bar\\', 'baz\\']))
|
||||
['foo']
|
||||
"""
|
||||
lines = iter(lines)
|
||||
for item in lines:
|
||||
while item.endswith('\\'):
|
||||
try:
|
||||
item = item[:-2].strip() + next(lines)
|
||||
except StopIteration:
|
||||
return
|
||||
yield item
|
||||
BIN
Binary file not shown.
@@ -1,4 +0,0 @@
|
||||
from .more import * # noqa
|
||||
from .recipes import * # noqa
|
||||
|
||||
__version__ = '8.8.0'
|
||||
BIN
Binary file not shown.
BIN
Binary file not shown.
BIN
Binary file not shown.
File diff suppressed because it is too large
Load Diff
@@ -1,620 +0,0 @@
|
||||
"""Imported from the recipes section of the itertools documentation.
|
||||
|
||||
All functions taken from the recipes section of the itertools library docs
|
||||
[1]_.
|
||||
Some backward-compatible usability improvements have been made.
|
||||
|
||||
.. [1] http://docs.python.org/library/itertools.html#recipes
|
||||
|
||||
"""
|
||||
import warnings
|
||||
from collections import deque
|
||||
from itertools import (
|
||||
chain,
|
||||
combinations,
|
||||
count,
|
||||
cycle,
|
||||
groupby,
|
||||
islice,
|
||||
repeat,
|
||||
starmap,
|
||||
tee,
|
||||
zip_longest,
|
||||
)
|
||||
import operator
|
||||
from random import randrange, sample, choice
|
||||
|
||||
__all__ = [
|
||||
'all_equal',
|
||||
'consume',
|
||||
'convolve',
|
||||
'dotproduct',
|
||||
'first_true',
|
||||
'flatten',
|
||||
'grouper',
|
||||
'iter_except',
|
||||
'ncycles',
|
||||
'nth',
|
||||
'nth_combination',
|
||||
'padnone',
|
||||
'pad_none',
|
||||
'pairwise',
|
||||
'partition',
|
||||
'powerset',
|
||||
'prepend',
|
||||
'quantify',
|
||||
'random_combination_with_replacement',
|
||||
'random_combination',
|
||||
'random_permutation',
|
||||
'random_product',
|
||||
'repeatfunc',
|
||||
'roundrobin',
|
||||
'tabulate',
|
||||
'tail',
|
||||
'take',
|
||||
'unique_everseen',
|
||||
'unique_justseen',
|
||||
]
|
||||
|
||||
|
||||
def take(n, iterable):
|
||||
"""Return first *n* items of the iterable as a list.
|
||||
|
||||
>>> take(3, range(10))
|
||||
[0, 1, 2]
|
||||
|
||||
If there are fewer than *n* items in the iterable, all of them are
|
||||
returned.
|
||||
|
||||
>>> take(10, range(3))
|
||||
[0, 1, 2]
|
||||
|
||||
"""
|
||||
return list(islice(iterable, n))
|
||||
|
||||
|
||||
def tabulate(function, start=0):
|
||||
"""Return an iterator over the results of ``func(start)``,
|
||||
``func(start + 1)``, ``func(start + 2)``...
|
||||
|
||||
*func* should be a function that accepts one integer argument.
|
||||
|
||||
If *start* is not specified it defaults to 0. It will be incremented each
|
||||
time the iterator is advanced.
|
||||
|
||||
>>> square = lambda x: x ** 2
|
||||
>>> iterator = tabulate(square, -3)
|
||||
>>> take(4, iterator)
|
||||
[9, 4, 1, 0]
|
||||
|
||||
"""
|
||||
return map(function, count(start))
|
||||
|
||||
|
||||
def tail(n, iterable):
|
||||
"""Return an iterator over the last *n* items of *iterable*.
|
||||
|
||||
>>> t = tail(3, 'ABCDEFG')
|
||||
>>> list(t)
|
||||
['E', 'F', 'G']
|
||||
|
||||
"""
|
||||
return iter(deque(iterable, maxlen=n))
|
||||
|
||||
|
||||
def consume(iterator, n=None):
|
||||
"""Advance *iterable* by *n* steps. If *n* is ``None``, consume it
|
||||
entirely.
|
||||
|
||||
Efficiently exhausts an iterator without returning values. Defaults to
|
||||
consuming the whole iterator, but an optional second argument may be
|
||||
provided to limit consumption.
|
||||
|
||||
>>> i = (x for x in range(10))
|
||||
>>> next(i)
|
||||
0
|
||||
>>> consume(i, 3)
|
||||
>>> next(i)
|
||||
4
|
||||
>>> consume(i)
|
||||
>>> next(i)
|
||||
Traceback (most recent call last):
|
||||
File "<stdin>", line 1, in <module>
|
||||
StopIteration
|
||||
|
||||
If the iterator has fewer items remaining than the provided limit, the
|
||||
whole iterator will be consumed.
|
||||
|
||||
>>> i = (x for x in range(3))
|
||||
>>> consume(i, 5)
|
||||
>>> next(i)
|
||||
Traceback (most recent call last):
|
||||
File "<stdin>", line 1, in <module>
|
||||
StopIteration
|
||||
|
||||
"""
|
||||
# Use functions that consume iterators at C speed.
|
||||
if n is None:
|
||||
# feed the entire iterator into a zero-length deque
|
||||
deque(iterator, maxlen=0)
|
||||
else:
|
||||
# advance to the empty slice starting at position n
|
||||
next(islice(iterator, n, n), None)
|
||||
|
||||
|
||||
def nth(iterable, n, default=None):
|
||||
"""Returns the nth item or a default value.
|
||||
|
||||
>>> l = range(10)
|
||||
>>> nth(l, 3)
|
||||
3
|
||||
>>> nth(l, 20, "zebra")
|
||||
'zebra'
|
||||
|
||||
"""
|
||||
return next(islice(iterable, n, None), default)
|
||||
|
||||
|
||||
def all_equal(iterable):
|
||||
"""
|
||||
Returns ``True`` if all the elements are equal to each other.
|
||||
|
||||
>>> all_equal('aaaa')
|
||||
True
|
||||
>>> all_equal('aaab')
|
||||
False
|
||||
|
||||
"""
|
||||
g = groupby(iterable)
|
||||
return next(g, True) and not next(g, False)
|
||||
|
||||
|
||||
def quantify(iterable, pred=bool):
|
||||
"""Return the how many times the predicate is true.
|
||||
|
||||
>>> quantify([True, False, True])
|
||||
2
|
||||
|
||||
"""
|
||||
return sum(map(pred, iterable))
|
||||
|
||||
|
||||
def pad_none(iterable):
|
||||
"""Returns the sequence of elements and then returns ``None`` indefinitely.
|
||||
|
||||
>>> take(5, pad_none(range(3)))
|
||||
[0, 1, 2, None, None]
|
||||
|
||||
Useful for emulating the behavior of the built-in :func:`map` function.
|
||||
|
||||
See also :func:`padded`.
|
||||
|
||||
"""
|
||||
return chain(iterable, repeat(None))
|
||||
|
||||
|
||||
padnone = pad_none
|
||||
|
||||
|
||||
def ncycles(iterable, n):
|
||||
"""Returns the sequence elements *n* times
|
||||
|
||||
>>> list(ncycles(["a", "b"], 3))
|
||||
['a', 'b', 'a', 'b', 'a', 'b']
|
||||
|
||||
"""
|
||||
return chain.from_iterable(repeat(tuple(iterable), n))
|
||||
|
||||
|
||||
def dotproduct(vec1, vec2):
|
||||
"""Returns the dot product of the two iterables.
|
||||
|
||||
>>> dotproduct([10, 10], [20, 20])
|
||||
400
|
||||
|
||||
"""
|
||||
return sum(map(operator.mul, vec1, vec2))
|
||||
|
||||
|
||||
def flatten(listOfLists):
|
||||
"""Return an iterator flattening one level of nesting in a list of lists.
|
||||
|
||||
>>> list(flatten([[0, 1], [2, 3]]))
|
||||
[0, 1, 2, 3]
|
||||
|
||||
See also :func:`collapse`, which can flatten multiple levels of nesting.
|
||||
|
||||
"""
|
||||
return chain.from_iterable(listOfLists)
|
||||
|
||||
|
||||
def repeatfunc(func, times=None, *args):
|
||||
"""Call *func* with *args* repeatedly, returning an iterable over the
|
||||
results.
|
||||
|
||||
If *times* is specified, the iterable will terminate after that many
|
||||
repetitions:
|
||||
|
||||
>>> from operator import add
|
||||
>>> times = 4
|
||||
>>> args = 3, 5
|
||||
>>> list(repeatfunc(add, times, *args))
|
||||
[8, 8, 8, 8]
|
||||
|
||||
If *times* is ``None`` the iterable will not terminate:
|
||||
|
||||
>>> from random import randrange
|
||||
>>> times = None
|
||||
>>> args = 1, 11
|
||||
>>> take(6, repeatfunc(randrange, times, *args)) # doctest:+SKIP
|
||||
[2, 4, 8, 1, 8, 4]
|
||||
|
||||
"""
|
||||
if times is None:
|
||||
return starmap(func, repeat(args))
|
||||
return starmap(func, repeat(args, times))
|
||||
|
||||
|
||||
def _pairwise(iterable):
|
||||
"""Returns an iterator of paired items, overlapping, from the original
|
||||
|
||||
>>> take(4, pairwise(count()))
|
||||
[(0, 1), (1, 2), (2, 3), (3, 4)]
|
||||
|
||||
On Python 3.10 and above, this is an alias for :func:`itertools.pairwise`.
|
||||
|
||||
"""
|
||||
a, b = tee(iterable)
|
||||
next(b, None)
|
||||
yield from zip(a, b)
|
||||
|
||||
|
||||
try:
|
||||
from itertools import pairwise as itertools_pairwise
|
||||
except ImportError:
|
||||
pairwise = _pairwise
|
||||
else:
|
||||
|
||||
def pairwise(iterable):
|
||||
yield from itertools_pairwise(iterable)
|
||||
|
||||
pairwise.__doc__ = _pairwise.__doc__
|
||||
|
||||
|
||||
def grouper(iterable, n, fillvalue=None):
|
||||
"""Collect data into fixed-length chunks or blocks.
|
||||
|
||||
>>> list(grouper('ABCDEFG', 3, 'x'))
|
||||
[('A', 'B', 'C'), ('D', 'E', 'F'), ('G', 'x', 'x')]
|
||||
|
||||
"""
|
||||
if isinstance(iterable, int):
|
||||
warnings.warn(
|
||||
"grouper expects iterable as first parameter", DeprecationWarning
|
||||
)
|
||||
n, iterable = iterable, n
|
||||
args = [iter(iterable)] * n
|
||||
return zip_longest(fillvalue=fillvalue, *args)
|
||||
|
||||
|
||||
def roundrobin(*iterables):
|
||||
"""Yields an item from each iterable, alternating between them.
|
||||
|
||||
>>> list(roundrobin('ABC', 'D', 'EF'))
|
||||
['A', 'D', 'E', 'B', 'F', 'C']
|
||||
|
||||
This function produces the same output as :func:`interleave_longest`, but
|
||||
may perform better for some inputs (in particular when the number of
|
||||
iterables is small).
|
||||
|
||||
"""
|
||||
# Recipe credited to George Sakkis
|
||||
pending = len(iterables)
|
||||
nexts = cycle(iter(it).__next__ for it in iterables)
|
||||
while pending:
|
||||
try:
|
||||
for next in nexts:
|
||||
yield next()
|
||||
except StopIteration:
|
||||
pending -= 1
|
||||
nexts = cycle(islice(nexts, pending))
|
||||
|
||||
|
||||
def partition(pred, iterable):
|
||||
"""
|
||||
Returns a 2-tuple of iterables derived from the input iterable.
|
||||
The first yields the items that have ``pred(item) == False``.
|
||||
The second yields the items that have ``pred(item) == True``.
|
||||
|
||||
>>> is_odd = lambda x: x % 2 != 0
|
||||
>>> iterable = range(10)
|
||||
>>> even_items, odd_items = partition(is_odd, iterable)
|
||||
>>> list(even_items), list(odd_items)
|
||||
([0, 2, 4, 6, 8], [1, 3, 5, 7, 9])
|
||||
|
||||
If *pred* is None, :func:`bool` is used.
|
||||
|
||||
>>> iterable = [0, 1, False, True, '', ' ']
|
||||
>>> false_items, true_items = partition(None, iterable)
|
||||
>>> list(false_items), list(true_items)
|
||||
([0, False, ''], [1, True, ' '])
|
||||
|
||||
"""
|
||||
if pred is None:
|
||||
pred = bool
|
||||
|
||||
evaluations = ((pred(x), x) for x in iterable)
|
||||
t1, t2 = tee(evaluations)
|
||||
return (
|
||||
(x for (cond, x) in t1 if not cond),
|
||||
(x for (cond, x) in t2 if cond),
|
||||
)
|
||||
|
||||
|
||||
def powerset(iterable):
|
||||
"""Yields all possible subsets of the iterable.
|
||||
|
||||
>>> list(powerset([1, 2, 3]))
|
||||
[(), (1,), (2,), (3,), (1, 2), (1, 3), (2, 3), (1, 2, 3)]
|
||||
|
||||
:func:`powerset` will operate on iterables that aren't :class:`set`
|
||||
instances, so repeated elements in the input will produce repeated elements
|
||||
in the output. Use :func:`unique_everseen` on the input to avoid generating
|
||||
duplicates:
|
||||
|
||||
>>> seq = [1, 1, 0]
|
||||
>>> list(powerset(seq))
|
||||
[(), (1,), (1,), (0,), (1, 1), (1, 0), (1, 0), (1, 1, 0)]
|
||||
>>> from more_itertools import unique_everseen
|
||||
>>> list(powerset(unique_everseen(seq)))
|
||||
[(), (1,), (0,), (1, 0)]
|
||||
|
||||
"""
|
||||
s = list(iterable)
|
||||
return chain.from_iterable(combinations(s, r) for r in range(len(s) + 1))
|
||||
|
||||
|
||||
def unique_everseen(iterable, key=None):
|
||||
"""
|
||||
Yield unique elements, preserving order.
|
||||
|
||||
>>> list(unique_everseen('AAAABBBCCDAABBB'))
|
||||
['A', 'B', 'C', 'D']
|
||||
>>> list(unique_everseen('ABBCcAD', str.lower))
|
||||
['A', 'B', 'C', 'D']
|
||||
|
||||
Sequences with a mix of hashable and unhashable items can be used.
|
||||
The function will be slower (i.e., `O(n^2)`) for unhashable items.
|
||||
|
||||
Remember that ``list`` objects are unhashable - you can use the *key*
|
||||
parameter to transform the list to a tuple (which is hashable) to
|
||||
avoid a slowdown.
|
||||
|
||||
>>> iterable = ([1, 2], [2, 3], [1, 2])
|
||||
>>> list(unique_everseen(iterable)) # Slow
|
||||
[[1, 2], [2, 3]]
|
||||
>>> list(unique_everseen(iterable, key=tuple)) # Faster
|
||||
[[1, 2], [2, 3]]
|
||||
|
||||
Similary, you may want to convert unhashable ``set`` objects with
|
||||
``key=frozenset``. For ``dict`` objects,
|
||||
``key=lambda x: frozenset(x.items())`` can be used.
|
||||
|
||||
"""
|
||||
seenset = set()
|
||||
seenset_add = seenset.add
|
||||
seenlist = []
|
||||
seenlist_add = seenlist.append
|
||||
use_key = key is not None
|
||||
|
||||
for element in iterable:
|
||||
k = key(element) if use_key else element
|
||||
try:
|
||||
if k not in seenset:
|
||||
seenset_add(k)
|
||||
yield element
|
||||
except TypeError:
|
||||
if k not in seenlist:
|
||||
seenlist_add(k)
|
||||
yield element
|
||||
|
||||
|
||||
def unique_justseen(iterable, key=None):
|
||||
"""Yields elements in order, ignoring serial duplicates
|
||||
|
||||
>>> list(unique_justseen('AAAABBBCCDAABBB'))
|
||||
['A', 'B', 'C', 'D', 'A', 'B']
|
||||
>>> list(unique_justseen('ABBCcAD', str.lower))
|
||||
['A', 'B', 'C', 'A', 'D']
|
||||
|
||||
"""
|
||||
return map(next, map(operator.itemgetter(1), groupby(iterable, key)))
|
||||
|
||||
|
||||
def iter_except(func, exception, first=None):
|
||||
"""Yields results from a function repeatedly until an exception is raised.
|
||||
|
||||
Converts a call-until-exception interface to an iterator interface.
|
||||
Like ``iter(func, sentinel)``, but uses an exception instead of a sentinel
|
||||
to end the loop.
|
||||
|
||||
>>> l = [0, 1, 2]
|
||||
>>> list(iter_except(l.pop, IndexError))
|
||||
[2, 1, 0]
|
||||
|
||||
"""
|
||||
try:
|
||||
if first is not None:
|
||||
yield first()
|
||||
while 1:
|
||||
yield func()
|
||||
except exception:
|
||||
pass
|
||||
|
||||
|
||||
def first_true(iterable, default=None, pred=None):
|
||||
"""
|
||||
Returns the first true value in the iterable.
|
||||
|
||||
If no true value is found, returns *default*
|
||||
|
||||
If *pred* is not None, returns the first item for which
|
||||
``pred(item) == True`` .
|
||||
|
||||
>>> first_true(range(10))
|
||||
1
|
||||
>>> first_true(range(10), pred=lambda x: x > 5)
|
||||
6
|
||||
>>> first_true(range(10), default='missing', pred=lambda x: x > 9)
|
||||
'missing'
|
||||
|
||||
"""
|
||||
return next(filter(pred, iterable), default)
|
||||
|
||||
|
||||
def random_product(*args, repeat=1):
|
||||
"""Draw an item at random from each of the input iterables.
|
||||
|
||||
>>> random_product('abc', range(4), 'XYZ') # doctest:+SKIP
|
||||
('c', 3, 'Z')
|
||||
|
||||
If *repeat* is provided as a keyword argument, that many items will be
|
||||
drawn from each iterable.
|
||||
|
||||
>>> random_product('abcd', range(4), repeat=2) # doctest:+SKIP
|
||||
('a', 2, 'd', 3)
|
||||
|
||||
This equivalent to taking a random selection from
|
||||
``itertools.product(*args, **kwarg)``.
|
||||
|
||||
"""
|
||||
pools = [tuple(pool) for pool in args] * repeat
|
||||
return tuple(choice(pool) for pool in pools)
|
||||
|
||||
|
||||
def random_permutation(iterable, r=None):
|
||||
"""Return a random *r* length permutation of the elements in *iterable*.
|
||||
|
||||
If *r* is not specified or is ``None``, then *r* defaults to the length of
|
||||
*iterable*.
|
||||
|
||||
>>> random_permutation(range(5)) # doctest:+SKIP
|
||||
(3, 4, 0, 1, 2)
|
||||
|
||||
This equivalent to taking a random selection from
|
||||
``itertools.permutations(iterable, r)``.
|
||||
|
||||
"""
|
||||
pool = tuple(iterable)
|
||||
r = len(pool) if r is None else r
|
||||
return tuple(sample(pool, r))
|
||||
|
||||
|
||||
def random_combination(iterable, r):
|
||||
"""Return a random *r* length subsequence of the elements in *iterable*.
|
||||
|
||||
>>> random_combination(range(5), 3) # doctest:+SKIP
|
||||
(2, 3, 4)
|
||||
|
||||
This equivalent to taking a random selection from
|
||||
``itertools.combinations(iterable, r)``.
|
||||
|
||||
"""
|
||||
pool = tuple(iterable)
|
||||
n = len(pool)
|
||||
indices = sorted(sample(range(n), r))
|
||||
return tuple(pool[i] for i in indices)
|
||||
|
||||
|
||||
def random_combination_with_replacement(iterable, r):
|
||||
"""Return a random *r* length subsequence of elements in *iterable*,
|
||||
allowing individual elements to be repeated.
|
||||
|
||||
>>> random_combination_with_replacement(range(3), 5) # doctest:+SKIP
|
||||
(0, 0, 1, 2, 2)
|
||||
|
||||
This equivalent to taking a random selection from
|
||||
``itertools.combinations_with_replacement(iterable, r)``.
|
||||
|
||||
"""
|
||||
pool = tuple(iterable)
|
||||
n = len(pool)
|
||||
indices = sorted(randrange(n) for i in range(r))
|
||||
return tuple(pool[i] for i in indices)
|
||||
|
||||
|
||||
def nth_combination(iterable, r, index):
|
||||
"""Equivalent to ``list(combinations(iterable, r))[index]``.
|
||||
|
||||
The subsequences of *iterable* that are of length *r* can be ordered
|
||||
lexicographically. :func:`nth_combination` computes the subsequence at
|
||||
sort position *index* directly, without computing the previous
|
||||
subsequences.
|
||||
|
||||
>>> nth_combination(range(5), 3, 5)
|
||||
(0, 3, 4)
|
||||
|
||||
``ValueError`` will be raised If *r* is negative or greater than the length
|
||||
of *iterable*.
|
||||
``IndexError`` will be raised if the given *index* is invalid.
|
||||
"""
|
||||
pool = tuple(iterable)
|
||||
n = len(pool)
|
||||
if (r < 0) or (r > n):
|
||||
raise ValueError
|
||||
|
||||
c = 1
|
||||
k = min(r, n - r)
|
||||
for i in range(1, k + 1):
|
||||
c = c * (n - k + i) // i
|
||||
|
||||
if index < 0:
|
||||
index += c
|
||||
|
||||
if (index < 0) or (index >= c):
|
||||
raise IndexError
|
||||
|
||||
result = []
|
||||
while r:
|
||||
c, n, r = c * r // n, n - 1, r - 1
|
||||
while index >= c:
|
||||
index -= c
|
||||
c, n = c * (n - r) // n, n - 1
|
||||
result.append(pool[-1 - n])
|
||||
|
||||
return tuple(result)
|
||||
|
||||
|
||||
def prepend(value, iterator):
|
||||
"""Yield *value*, followed by the elements in *iterator*.
|
||||
|
||||
>>> value = '0'
|
||||
>>> iterator = ['1', '2', '3']
|
||||
>>> list(prepend(value, iterator))
|
||||
['0', '1', '2', '3']
|
||||
|
||||
To prepend multiple values, see :func:`itertools.chain`
|
||||
or :func:`value_chain`.
|
||||
|
||||
"""
|
||||
return chain([value], iterator)
|
||||
|
||||
|
||||
def convolve(signal, kernel):
|
||||
"""Convolve the iterable *signal* with the iterable *kernel*.
|
||||
|
||||
>>> signal = (1, 2, 3, 4, 5)
|
||||
>>> kernel = [3, 2, 1]
|
||||
>>> list(convolve(signal, kernel))
|
||||
[3, 8, 14, 20, 26, 14, 5]
|
||||
|
||||
Note: the input arguments are not interchangeable, as the *kernel*
|
||||
is immediately consumed and stored.
|
||||
|
||||
"""
|
||||
kernel = tuple(kernel)[::-1]
|
||||
n = len(kernel)
|
||||
window = deque([0], maxlen=n) * n
|
||||
for x in chain(signal, repeat(0, n - 1)):
|
||||
window.append(x)
|
||||
yield sum(map(operator.mul, kernel, window))
|
||||
@@ -1,488 +0,0 @@
|
||||
"""
|
||||
An OrderedSet is a custom MutableSet that remembers its order, so that every
|
||||
entry has an index that can be looked up.
|
||||
|
||||
Based on a recipe originally posted to ActiveState Recipes by Raymond Hettiger,
|
||||
and released under the MIT license.
|
||||
"""
|
||||
import itertools as it
|
||||
from collections import deque
|
||||
|
||||
try:
|
||||
# Python 3
|
||||
from collections.abc import MutableSet, Sequence
|
||||
except ImportError:
|
||||
# Python 2.7
|
||||
from collections import MutableSet, Sequence
|
||||
|
||||
SLICE_ALL = slice(None)
|
||||
__version__ = "3.1"
|
||||
|
||||
|
||||
def is_iterable(obj):
|
||||
"""
|
||||
Are we being asked to look up a list of things, instead of a single thing?
|
||||
We check for the `__iter__` attribute so that this can cover types that
|
||||
don't have to be known by this module, such as NumPy arrays.
|
||||
|
||||
Strings, however, should be considered as atomic values to look up, not
|
||||
iterables. The same goes for tuples, since they are immutable and therefore
|
||||
valid entries.
|
||||
|
||||
We don't need to check for the Python 2 `unicode` type, because it doesn't
|
||||
have an `__iter__` attribute anyway.
|
||||
"""
|
||||
return (
|
||||
hasattr(obj, "__iter__")
|
||||
and not isinstance(obj, str)
|
||||
and not isinstance(obj, tuple)
|
||||
)
|
||||
|
||||
|
||||
class OrderedSet(MutableSet, Sequence):
|
||||
"""
|
||||
An OrderedSet is a custom MutableSet that remembers its order, so that
|
||||
every entry has an index that can be looked up.
|
||||
|
||||
Example:
|
||||
>>> OrderedSet([1, 1, 2, 3, 2])
|
||||
OrderedSet([1, 2, 3])
|
||||
"""
|
||||
|
||||
def __init__(self, iterable=None):
|
||||
self.items = []
|
||||
self.map = {}
|
||||
if iterable is not None:
|
||||
self |= iterable
|
||||
|
||||
def __len__(self):
|
||||
"""
|
||||
Returns the number of unique elements in the ordered set
|
||||
|
||||
Example:
|
||||
>>> len(OrderedSet([]))
|
||||
0
|
||||
>>> len(OrderedSet([1, 2]))
|
||||
2
|
||||
"""
|
||||
return len(self.items)
|
||||
|
||||
def __getitem__(self, index):
|
||||
"""
|
||||
Get the item at a given index.
|
||||
|
||||
If `index` is a slice, you will get back that slice of items, as a
|
||||
new OrderedSet.
|
||||
|
||||
If `index` is a list or a similar iterable, you'll get a list of
|
||||
items corresponding to those indices. This is similar to NumPy's
|
||||
"fancy indexing". The result is not an OrderedSet because you may ask
|
||||
for duplicate indices, and the number of elements returned should be
|
||||
the number of elements asked for.
|
||||
|
||||
Example:
|
||||
>>> oset = OrderedSet([1, 2, 3])
|
||||
>>> oset[1]
|
||||
2
|
||||
"""
|
||||
if isinstance(index, slice) and index == SLICE_ALL:
|
||||
return self.copy()
|
||||
elif is_iterable(index):
|
||||
return [self.items[i] for i in index]
|
||||
elif hasattr(index, "__index__") or isinstance(index, slice):
|
||||
result = self.items[index]
|
||||
if isinstance(result, list):
|
||||
return self.__class__(result)
|
||||
else:
|
||||
return result
|
||||
else:
|
||||
raise TypeError("Don't know how to index an OrderedSet by %r" % index)
|
||||
|
||||
def copy(self):
|
||||
"""
|
||||
Return a shallow copy of this object.
|
||||
|
||||
Example:
|
||||
>>> this = OrderedSet([1, 2, 3])
|
||||
>>> other = this.copy()
|
||||
>>> this == other
|
||||
True
|
||||
>>> this is other
|
||||
False
|
||||
"""
|
||||
return self.__class__(self)
|
||||
|
||||
def __getstate__(self):
|
||||
if len(self) == 0:
|
||||
# The state can't be an empty list.
|
||||
# We need to return a truthy value, or else __setstate__ won't be run.
|
||||
#
|
||||
# This could have been done more gracefully by always putting the state
|
||||
# in a tuple, but this way is backwards- and forwards- compatible with
|
||||
# previous versions of OrderedSet.
|
||||
return (None,)
|
||||
else:
|
||||
return list(self)
|
||||
|
||||
def __setstate__(self, state):
|
||||
if state == (None,):
|
||||
self.__init__([])
|
||||
else:
|
||||
self.__init__(state)
|
||||
|
||||
def __contains__(self, key):
|
||||
"""
|
||||
Test if the item is in this ordered set
|
||||
|
||||
Example:
|
||||
>>> 1 in OrderedSet([1, 3, 2])
|
||||
True
|
||||
>>> 5 in OrderedSet([1, 3, 2])
|
||||
False
|
||||
"""
|
||||
return key in self.map
|
||||
|
||||
def add(self, key):
|
||||
"""
|
||||
Add `key` as an item to this OrderedSet, then return its index.
|
||||
|
||||
If `key` is already in the OrderedSet, return the index it already
|
||||
had.
|
||||
|
||||
Example:
|
||||
>>> oset = OrderedSet()
|
||||
>>> oset.append(3)
|
||||
0
|
||||
>>> print(oset)
|
||||
OrderedSet([3])
|
||||
"""
|
||||
if key not in self.map:
|
||||
self.map[key] = len(self.items)
|
||||
self.items.append(key)
|
||||
return self.map[key]
|
||||
|
||||
append = add
|
||||
|
||||
def update(self, sequence):
|
||||
"""
|
||||
Update the set with the given iterable sequence, then return the index
|
||||
of the last element inserted.
|
||||
|
||||
Example:
|
||||
>>> oset = OrderedSet([1, 2, 3])
|
||||
>>> oset.update([3, 1, 5, 1, 4])
|
||||
4
|
||||
>>> print(oset)
|
||||
OrderedSet([1, 2, 3, 5, 4])
|
||||
"""
|
||||
item_index = None
|
||||
try:
|
||||
for item in sequence:
|
||||
item_index = self.add(item)
|
||||
except TypeError:
|
||||
raise ValueError(
|
||||
"Argument needs to be an iterable, got %s" % type(sequence)
|
||||
)
|
||||
return item_index
|
||||
|
||||
def index(self, key):
|
||||
"""
|
||||
Get the index of a given entry, raising an IndexError if it's not
|
||||
present.
|
||||
|
||||
`key` can be an iterable of entries that is not a string, in which case
|
||||
this returns a list of indices.
|
||||
|
||||
Example:
|
||||
>>> oset = OrderedSet([1, 2, 3])
|
||||
>>> oset.index(2)
|
||||
1
|
||||
"""
|
||||
if is_iterable(key):
|
||||
return [self.index(subkey) for subkey in key]
|
||||
return self.map[key]
|
||||
|
||||
# Provide some compatibility with pd.Index
|
||||
get_loc = index
|
||||
get_indexer = index
|
||||
|
||||
def pop(self):
|
||||
"""
|
||||
Remove and return the last element from the set.
|
||||
|
||||
Raises KeyError if the set is empty.
|
||||
|
||||
Example:
|
||||
>>> oset = OrderedSet([1, 2, 3])
|
||||
>>> oset.pop()
|
||||
3
|
||||
"""
|
||||
if not self.items:
|
||||
raise KeyError("Set is empty")
|
||||
|
||||
elem = self.items[-1]
|
||||
del self.items[-1]
|
||||
del self.map[elem]
|
||||
return elem
|
||||
|
||||
def discard(self, key):
|
||||
"""
|
||||
Remove an element. Do not raise an exception if absent.
|
||||
|
||||
The MutableSet mixin uses this to implement the .remove() method, which
|
||||
*does* raise an error when asked to remove a non-existent item.
|
||||
|
||||
Example:
|
||||
>>> oset = OrderedSet([1, 2, 3])
|
||||
>>> oset.discard(2)
|
||||
>>> print(oset)
|
||||
OrderedSet([1, 3])
|
||||
>>> oset.discard(2)
|
||||
>>> print(oset)
|
||||
OrderedSet([1, 3])
|
||||
"""
|
||||
if key in self:
|
||||
i = self.map[key]
|
||||
del self.items[i]
|
||||
del self.map[key]
|
||||
for k, v in self.map.items():
|
||||
if v >= i:
|
||||
self.map[k] = v - 1
|
||||
|
||||
def clear(self):
|
||||
"""
|
||||
Remove all items from this OrderedSet.
|
||||
"""
|
||||
del self.items[:]
|
||||
self.map.clear()
|
||||
|
||||
def __iter__(self):
|
||||
"""
|
||||
Example:
|
||||
>>> list(iter(OrderedSet([1, 2, 3])))
|
||||
[1, 2, 3]
|
||||
"""
|
||||
return iter(self.items)
|
||||
|
||||
def __reversed__(self):
|
||||
"""
|
||||
Example:
|
||||
>>> list(reversed(OrderedSet([1, 2, 3])))
|
||||
[3, 2, 1]
|
||||
"""
|
||||
return reversed(self.items)
|
||||
|
||||
def __repr__(self):
|
||||
if not self:
|
||||
return "%s()" % (self.__class__.__name__,)
|
||||
return "%s(%r)" % (self.__class__.__name__, list(self))
|
||||
|
||||
def __eq__(self, other):
|
||||
"""
|
||||
Returns true if the containers have the same items. If `other` is a
|
||||
Sequence, then order is checked, otherwise it is ignored.
|
||||
|
||||
Example:
|
||||
>>> oset = OrderedSet([1, 3, 2])
|
||||
>>> oset == [1, 3, 2]
|
||||
True
|
||||
>>> oset == [1, 2, 3]
|
||||
False
|
||||
>>> oset == [2, 3]
|
||||
False
|
||||
>>> oset == OrderedSet([3, 2, 1])
|
||||
False
|
||||
"""
|
||||
# In Python 2 deque is not a Sequence, so treat it as one for
|
||||
# consistent behavior with Python 3.
|
||||
if isinstance(other, (Sequence, deque)):
|
||||
# Check that this OrderedSet contains the same elements, in the
|
||||
# same order, as the other object.
|
||||
return list(self) == list(other)
|
||||
try:
|
||||
other_as_set = set(other)
|
||||
except TypeError:
|
||||
# If `other` can't be converted into a set, it's not equal.
|
||||
return False
|
||||
else:
|
||||
return set(self) == other_as_set
|
||||
|
||||
def union(self, *sets):
|
||||
"""
|
||||
Combines all unique items.
|
||||
Each items order is defined by its first appearance.
|
||||
|
||||
Example:
|
||||
>>> oset = OrderedSet.union(OrderedSet([3, 1, 4, 1, 5]), [1, 3], [2, 0])
|
||||
>>> print(oset)
|
||||
OrderedSet([3, 1, 4, 5, 2, 0])
|
||||
>>> oset.union([8, 9])
|
||||
OrderedSet([3, 1, 4, 5, 2, 0, 8, 9])
|
||||
>>> oset | {10}
|
||||
OrderedSet([3, 1, 4, 5, 2, 0, 10])
|
||||
"""
|
||||
cls = self.__class__ if isinstance(self, OrderedSet) else OrderedSet
|
||||
containers = map(list, it.chain([self], sets))
|
||||
items = it.chain.from_iterable(containers)
|
||||
return cls(items)
|
||||
|
||||
def __and__(self, other):
|
||||
# the parent implementation of this is backwards
|
||||
return self.intersection(other)
|
||||
|
||||
def intersection(self, *sets):
|
||||
"""
|
||||
Returns elements in common between all sets. Order is defined only
|
||||
by the first set.
|
||||
|
||||
Example:
|
||||
>>> oset = OrderedSet.intersection(OrderedSet([0, 1, 2, 3]), [1, 2, 3])
|
||||
>>> print(oset)
|
||||
OrderedSet([1, 2, 3])
|
||||
>>> oset.intersection([2, 4, 5], [1, 2, 3, 4])
|
||||
OrderedSet([2])
|
||||
>>> oset.intersection()
|
||||
OrderedSet([1, 2, 3])
|
||||
"""
|
||||
cls = self.__class__ if isinstance(self, OrderedSet) else OrderedSet
|
||||
if sets:
|
||||
common = set.intersection(*map(set, sets))
|
||||
items = (item for item in self if item in common)
|
||||
else:
|
||||
items = self
|
||||
return cls(items)
|
||||
|
||||
def difference(self, *sets):
|
||||
"""
|
||||
Returns all elements that are in this set but not the others.
|
||||
|
||||
Example:
|
||||
>>> OrderedSet([1, 2, 3]).difference(OrderedSet([2]))
|
||||
OrderedSet([1, 3])
|
||||
>>> OrderedSet([1, 2, 3]).difference(OrderedSet([2]), OrderedSet([3]))
|
||||
OrderedSet([1])
|
||||
>>> OrderedSet([1, 2, 3]) - OrderedSet([2])
|
||||
OrderedSet([1, 3])
|
||||
>>> OrderedSet([1, 2, 3]).difference()
|
||||
OrderedSet([1, 2, 3])
|
||||
"""
|
||||
cls = self.__class__
|
||||
if sets:
|
||||
other = set.union(*map(set, sets))
|
||||
items = (item for item in self if item not in other)
|
||||
else:
|
||||
items = self
|
||||
return cls(items)
|
||||
|
||||
def issubset(self, other):
|
||||
"""
|
||||
Report whether another set contains this set.
|
||||
|
||||
Example:
|
||||
>>> OrderedSet([1, 2, 3]).issubset({1, 2})
|
||||
False
|
||||
>>> OrderedSet([1, 2, 3]).issubset({1, 2, 3, 4})
|
||||
True
|
||||
>>> OrderedSet([1, 2, 3]).issubset({1, 4, 3, 5})
|
||||
False
|
||||
"""
|
||||
if len(self) > len(other): # Fast check for obvious cases
|
||||
return False
|
||||
return all(item in other for item in self)
|
||||
|
||||
def issuperset(self, other):
|
||||
"""
|
||||
Report whether this set contains another set.
|
||||
|
||||
Example:
|
||||
>>> OrderedSet([1, 2]).issuperset([1, 2, 3])
|
||||
False
|
||||
>>> OrderedSet([1, 2, 3, 4]).issuperset({1, 2, 3})
|
||||
True
|
||||
>>> OrderedSet([1, 4, 3, 5]).issuperset({1, 2, 3})
|
||||
False
|
||||
"""
|
||||
if len(self) < len(other): # Fast check for obvious cases
|
||||
return False
|
||||
return all(item in self for item in other)
|
||||
|
||||
def symmetric_difference(self, other):
|
||||
"""
|
||||
Return the symmetric difference of two OrderedSets as a new set.
|
||||
That is, the new set will contain all elements that are in exactly
|
||||
one of the sets.
|
||||
|
||||
Their order will be preserved, with elements from `self` preceding
|
||||
elements from `other`.
|
||||
|
||||
Example:
|
||||
>>> this = OrderedSet([1, 4, 3, 5, 7])
|
||||
>>> other = OrderedSet([9, 7, 1, 3, 2])
|
||||
>>> this.symmetric_difference(other)
|
||||
OrderedSet([4, 5, 9, 2])
|
||||
"""
|
||||
cls = self.__class__ if isinstance(self, OrderedSet) else OrderedSet
|
||||
diff1 = cls(self).difference(other)
|
||||
diff2 = cls(other).difference(self)
|
||||
return diff1.union(diff2)
|
||||
|
||||
def _update_items(self, items):
|
||||
"""
|
||||
Replace the 'items' list of this OrderedSet with a new one, updating
|
||||
self.map accordingly.
|
||||
"""
|
||||
self.items = items
|
||||
self.map = {item: idx for (idx, item) in enumerate(items)}
|
||||
|
||||
def difference_update(self, *sets):
|
||||
"""
|
||||
Update this OrderedSet to remove items from one or more other sets.
|
||||
|
||||
Example:
|
||||
>>> this = OrderedSet([1, 2, 3])
|
||||
>>> this.difference_update(OrderedSet([2, 4]))
|
||||
>>> print(this)
|
||||
OrderedSet([1, 3])
|
||||
|
||||
>>> this = OrderedSet([1, 2, 3, 4, 5])
|
||||
>>> this.difference_update(OrderedSet([2, 4]), OrderedSet([1, 4, 6]))
|
||||
>>> print(this)
|
||||
OrderedSet([3, 5])
|
||||
"""
|
||||
items_to_remove = set()
|
||||
for other in sets:
|
||||
items_to_remove |= set(other)
|
||||
self._update_items([item for item in self.items if item not in items_to_remove])
|
||||
|
||||
def intersection_update(self, other):
|
||||
"""
|
||||
Update this OrderedSet to keep only items in another set, preserving
|
||||
their order in this set.
|
||||
|
||||
Example:
|
||||
>>> this = OrderedSet([1, 4, 3, 5, 7])
|
||||
>>> other = OrderedSet([9, 7, 1, 3, 2])
|
||||
>>> this.intersection_update(other)
|
||||
>>> print(this)
|
||||
OrderedSet([1, 3, 7])
|
||||
"""
|
||||
other = set(other)
|
||||
self._update_items([item for item in self.items if item in other])
|
||||
|
||||
def symmetric_difference_update(self, other):
|
||||
"""
|
||||
Update this OrderedSet to remove items from another set, then
|
||||
add items from the other set that were not present in this set.
|
||||
|
||||
Example:
|
||||
>>> this = OrderedSet([1, 4, 3, 5, 7])
|
||||
>>> other = OrderedSet([9, 7, 1, 3, 2])
|
||||
>>> this.symmetric_difference_update(other)
|
||||
>>> print(this)
|
||||
OrderedSet([4, 5, 9, 2])
|
||||
"""
|
||||
items_to_add = [item for item in other if item not in self]
|
||||
items_to_remove = set(other)
|
||||
self._update_items(
|
||||
[item for item in self.items if item not in items_to_remove] + items_to_add
|
||||
)
|
||||
@@ -1,26 +0,0 @@
|
||||
# This file is dual licensed under the terms of the Apache License, Version
|
||||
# 2.0, and the BSD License. See the LICENSE file in the root of this repository
|
||||
# for complete details.
|
||||
|
||||
__all__ = [
|
||||
"__title__",
|
||||
"__summary__",
|
||||
"__uri__",
|
||||
"__version__",
|
||||
"__author__",
|
||||
"__email__",
|
||||
"__license__",
|
||||
"__copyright__",
|
||||
]
|
||||
|
||||
__title__ = "packaging"
|
||||
__summary__ = "Core utilities for Python packages"
|
||||
__uri__ = "https://github.com/pypa/packaging"
|
||||
|
||||
__version__ = "21.3"
|
||||
|
||||
__author__ = "Donald Stufft and individual contributors"
|
||||
__email__ = "donald@stufft.io"
|
||||
|
||||
__license__ = "BSD-2-Clause or Apache-2.0"
|
||||
__copyright__ = "2014-2019 %s" % __author__
|
||||
@@ -1,25 +0,0 @@
|
||||
# This file is dual licensed under the terms of the Apache License, Version
|
||||
# 2.0, and the BSD License. See the LICENSE file in the root of this repository
|
||||
# for complete details.
|
||||
|
||||
from .__about__ import (
|
||||
__author__,
|
||||
__copyright__,
|
||||
__email__,
|
||||
__license__,
|
||||
__summary__,
|
||||
__title__,
|
||||
__uri__,
|
||||
__version__,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"__title__",
|
||||
"__summary__",
|
||||
"__uri__",
|
||||
"__version__",
|
||||
"__author__",
|
||||
"__email__",
|
||||
"__license__",
|
||||
"__copyright__",
|
||||
]
|
||||
BIN
Binary file not shown.
Reference in New Issue
Block a user