wrapped,

assigned = WRAPPER_ASSIGNMENTS,

updated = WRAPPER_UPDATES):

"""Update a wrapper function to look like the wrapped function

for attr in assigned:

try:

value = getattr(wrapped, attr)

except AttributeError:

pass

else:

setattr(wrapper, attr, value)

for attr in updated:

getattr(wrapper, attr).update(getattr(wrapped, attr, {}))

# Issue #17482: set __wrapped__ last so we don't inadvertently copy it

# from the wrapped function when updating __dict__

wrapper.__wrapped__ = wrapped

# Return the wrapper so this can be used as a decorator via partial()

assigned = WRAPPER_ASSIGNMENTS,

updated = WRAPPER_UPDATES):

"""Decorator factory to apply update_wrapper() to a wrapper function

return partial(update_wrapper, wrapped=wrapped,

'Return a > b. Computed by @total_ordering from (not a < b) and (a != b).'

op_result = type(self).__lt__(self, other)

if op_result is NotImplemented:

return op_result

'Return a <= b. Computed by @total_ordering from (a < b) or (a == b).'

op_result = type(self).__lt__(self, other)

if op_result is NotImplemented:

return op_result

'Return a >= b. Computed by @total_ordering from (not a < b).'

op_result = type(self).__lt__(self, other)

if op_result is NotImplemented:

return op_result

'Return a >= b. Computed by @total_ordering from (not a <= b) or (a == b).'

op_result = type(self).__le__(self, other)

if op_result is NotImplemented:

return op_result

'Return a < b. Computed by @total_ordering from (a <= b) and (a != b).'

op_result = type(self).__le__(self, other)

if op_result is NotImplemented:

return op_result

'Return a > b. Computed by @total_ordering from (not a <= b).'

op_result = type(self).__le__(self, other)

if op_result is NotImplemented:

return op_result

'Return a < b. Computed by @total_ordering from (not a > b) and (a != b).'

op_result = type(self).__gt__(self, other)

if op_result is NotImplemented:

return op_result

'Return a >= b. Computed by @total_ordering from (a > b) or (a == b).'

op_result = type(self).__gt__(self, other)

if op_result is NotImplemented:

return op_result

'Return a <= b. Computed by @total_ordering from (not a > b).'

op_result = type(self).__gt__(self, other)

if op_result is NotImplemented:

return op_result

'Return a <= b. Computed by @total_ordering from (not a >= b) or (a == b).'

op_result = type(self).__ge__(self, other)

if op_result is NotImplemented:

return op_result

'Return a > b. Computed by @total_ordering from (a >= b) and (a != b).'

op_result = type(self).__ge__(self, other)

if op_result is NotImplemented:

return op_result

'Return a < b. Computed by @total_ordering from (not a >= b).'

op_result = type(self).__ge__(self, other)

if op_result is NotImplemented:

return op_result

'__lt__': [('__gt__', _gt_from_lt),

('__le__', _le_from_lt),

('__ge__', _ge_from_lt)],

'__le__': [('__ge__', _ge_from_le),

('__lt__', _lt_from_le),

('__gt__', _gt_from_le)],

'__gt__': [('__lt__', _lt_from_gt),

('__ge__', _ge_from_gt),

('__le__', _le_from_gt)],

'__ge__': [('__le__', _le_from_ge),

('__gt__', _gt_from_ge),

('__lt__', _lt_from_ge)]

"""Class decorator that fills in missing ordering methods"""

# Find user-defined comparisons (not those inherited from object).

roots = {op for op in _convert if getattr(cls, op, None) is not getattr(object, op, None)}

if not roots:

raise ValueError('must define at least one ordering operation: < > <= >=')

root = max(roots) # prefer __lt__ to __le__ to __gt__ to __ge__

for opname, opfunc in _convert[root]:

if opname not in roots:

opfunc.__name__ = opname

setattr(cls, opname, opfunc)

"""Convert a cmp= function into a key= function"""

class K(object):

__slots__ = ['obj']

def __init__(self, obj):

self.obj = obj

def __lt__(self, other):

return mycmp(self.obj, other.obj) < 0

def __gt__(self, other):

return mycmp(self.obj, other.obj) > 0

def __eq__(self, other):

return mycmp(self.obj, other.obj) == 0

def __le__(self, other):

return mycmp(self.obj, other.obj) <= 0

def __ge__(self, other):

return mycmp(self.obj, other.obj) >= 0

__hash__ = None

"""

it = iter(sequence)

if initial is _initial_missing:

try:

value = next(it)

except StopIteration:

raise TypeError(

"reduce() of empty iterable with no initial value") from None

else:

value = initial

for element in it:

value = function(value, element)

"""New function with partial application of the given arguments

__slots__ = "func", "args", "keywords", "__dict__", "__weakref__"

def __new__(cls, func, /, *args, **keywords):

if not callable(func):

raise TypeError("the first argument must be callable")

if hasattr(func, "func"):

args = func.args + args

keywords = {**func.keywords, **keywords}

func = func.func

self = super(partial, cls).__new__(cls)

self.func = func

self.args = args

self.keywords = keywords

return self

def __call__(self, /, *args, **keywords):

keywords = {**self.keywords, **keywords}

return self.func(*self.args, *args, **keywords)

@recursive_repr()

def __repr__(self):

qualname = type(self).__qualname__

args = [repr(self.func)]

args.extend(repr(x) for x in self.args)

args.extend(f"{k}={v!r}" for (k, v) in self.keywords.items())

if type(self).__module__ == "functools":

return f"functools.{qualname}({', '.join(args)})"

return f"{qualname}({', '.join(args)})"

def __reduce__(self):

return type(self), (self.func,), (self.func, self.args,

self.keywords or None, self.__dict__ or None)

def __setstate__(self, state):

if not isinstance(state, tuple):

raise TypeError("argument to __setstate__ must be a tuple")

if len(state) != 4:

raise TypeError(f"expected 4 items in state, got {len(state)}")

func, args, kwds, namespace = state

if (not callable(func) or not isinstance(args, tuple) or

(kwds is not None and not isinstance(kwds, dict)) or

(namespace is not None and not isinstance(namespace, dict))):

raise TypeError("invalid partial state")

args = tuple(args) # just in case it's a subclass

if kwds is None:

kwds = {}

elif type(kwds) is not dict: # XXX does it need to be *exactly* dict?

kwds = dict(kwds)

if namespace is None:

namespace = {}

self.__dict__ = namespace

self.func = func

self.args = args

"""Method descriptor with partial application of the given arguments

def __init__(self, func, /, *args, **keywords):

if not callable(func) and not hasattr(func, "__get__"):

raise TypeError("{!r} is not callable or a descriptor"

.format(func))

# func could be a descriptor like classmethod which isn't callable,

# so we can't inherit from partial (it verifies func is callable)

if isinstance(func, partialmethod):

# flattening is mandatory in order to place cls/self before all

# other arguments

# it's also more efficient since only one function will be called

self.func = func.func

self.args = func.args + args

self.keywords = {**func.keywords, **keywords}

else:

self.func = func

self.args = args

self.keywords = keywords

def __repr__(self):

args = ", ".join(map(repr, self.args))

keywords = ", ".join("{}={!r}".format(k, v)

for k, v in self.keywords.items())

format_string = "{module}.{cls}({func}, {args}, {keywords})"

return format_string.format(module=self.__class__.__module__,

cls=self.__class__.__qualname__,

func=self.func,

args=args,

keywords=keywords)

def _make_unbound_method(self):

def _method(cls_or_self, /, *args, **keywords):

keywords = {**self.keywords, **keywords}

return self.func(cls_or_self, *self.args, *args, **keywords)

_method.__isabstractmethod__ = self.__isabstractmethod__

_method._partialmethod = self

return _method

def __get__(self, obj, cls=None):

get = getattr(self.func, "__get__", None)

result = None

if get is not None:

new_func = get(obj, cls)

if new_func is not self.func:

# Assume __get__ returning something new indicates the

# creation of an appropriate callable

result = partial(new_func, *self.args, **self.keywords)

try:

result.__self__ = new_func.__self__

except AttributeError:

pass

if result is None:

# If the underlying descriptor didn't do anything, treat this

# like an instance method

result = self._make_unbound_method().__get__(obj, cls)

return result

@property

def __isabstractmethod__(self):

return getattr(self.func, "__isabstractmethod__", False)

""" This class guarantees that hash() will be called no more than once

__slots__ = 'hashvalue'

def __init__(self, tup, hash=hash):

self[:] = tup

self.hashvalue = hash(tup)

def __hash__(self):

kwd_mark = (object(),),

fasttypes = {int, str},

tuple=tuple, type=type, len=len):

"""Make a cache key from optionally typed positional and keyword arguments

# All of code below relies on kwds preserving the order input by the user.

# Formerly, we sorted() the kwds before looping. The new way is *much*

# faster; however, it means that f(x=1, y=2) will now be treated as a

# distinct call from f(y=2, x=1) which will be cached separately.

key = args

if kwds:

key += kwd_mark

for item in kwds.items():

key += item

if typed:

key += tuple(type(v) for v in args)

if kwds:

key += tuple(type(v) for v in kwds.values())

elif len(key) == 1 and type(key[0]) in fasttypes:

return key[0]

"""Least-recently-used cache decorator.

# Users should only access the lru_cache through its public API:

# cache_info, cache_clear, and f.__wrapped__

# The internals of the lru_cache are encapsulated for thread safety and

# to allow the implementation to change (including a possible C version).

if isinstance(maxsize, int):

# Negative maxsize is treated as 0

if maxsize < 0:

maxsize = 0

elif callable(maxsize) and isinstance(typed, bool):

# The user_function was passed in directly via the maxsize argument

user_function, maxsize = maxsize, 128

wrapper = _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo)

wrapper.cache_parameters = lambda : {'maxsize': maxsize, 'typed': typed}

return update_wrapper(wrapper, user_function)

elif maxsize is not None:

raise TypeError(

'Expected first argument to be an integer, a callable, or None')

def decorating_function(user_function):

wrapper = _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo)

wrapper.cache_parameters = lambda : {'maxsize': maxsize, 'typed': typed}

return update_wrapper(wrapper, user_function)

# Constants shared by all lru cache instances:

sentinel = object() # unique object used to signal cache misses

make_key = _make_key # build a key from the function arguments

PREV, NEXT, KEY, RESULT = 0, 1, 2, 3 # names for the link fields

cache = {}

hits = misses = 0

full = False

cache_get = cache.get # bound method to lookup a key or return None

cache_len = cache.__len__ # get cache size without calling len()

lock = RLock() # because linkedlist updates aren't threadsafe

root = [] # root of the circular doubly linked list

root[:] = [root, root, None, None] # initialize by pointing to self

if maxsize == 0:

def wrapper(*args, **kwds):

# No caching -- just a statistics update

nonlocal misses

misses += 1

result = user_function(*args, **kwds)

return result

elif maxsize is None:

def wrapper(*args, **kwds):

# Simple caching without ordering or size limit

nonlocal hits, misses

key = make_key(args, kwds, typed)

result = cache_get(key, sentinel)

if result is not sentinel:

hits += 1

return result

misses += 1

result = user_function(*args, **kwds)

cache[key] = result

return result

else:

def wrapper(*args, **kwds):

# Size limited caching that tracks accesses by recency

nonlocal root, hits, misses, full

key = make_key(args, kwds, typed)

with lock:

link = cache_get(key)

if link is not None:

# Move the link to the front of the circular queue

link_prev, link_next, _key, result = link

link_prev[NEXT] = link_next

link_next[PREV] = link_prev

last = root[PREV]

last[NEXT] = root[PREV] = link

link[PREV] = last

link[NEXT] = root

hits += 1

return result

misses += 1

result = user_function(*args, **kwds)

with lock:

if key in cache:

# Getting here means that this same key was added to the

# cache while the lock was released. Since the link

# update is already done, we need only return the

# computed result and update the count of misses.

pass

elif full:

# Use the old root to store the new key and result.

oldroot = root

oldroot[KEY] = key

oldroot[RESULT] = result

# Empty the oldest link and make it the new root.

# Keep a reference to the old key and old result to

# prevent their ref counts from going to zero during the

# update. That will prevent potentially arbitrary object

# clean-up code (i.e. __del__) from running while we're

# still adjusting the links.

root = oldroot[NEXT]

oldkey = root[KEY]

oldresult = root[RESULT]

root[KEY] = root[RESULT] = None

# Now update the cache dictionary.

del cache[oldkey]

# Save the potentially reentrant cache[key] assignment

# for last, after the root and links have been put in

# a consistent state.

cache[key] = oldroot

else:

# Put result in a new link at the front of the queue.

last = root[PREV]

link = [last, root, key, result]

last[NEXT] = root[PREV] = cache[key] = link

# Use the cache_len bound method instead of the len() function

# which could potentially be wrapped in an lru_cache itself.

full = (cache_len() >= maxsize)

return result

def cache_info():

"""Report cache statistics"""

with lock:

return _CacheInfo(hits, misses, maxsize, cache_len())

def cache_clear():

"""Clear the cache and cache statistics"""

nonlocal hits, misses, full

with lock:

cache.clear()

root[:] = [root, root, None, None]

hits = misses = 0

full = False

wrapper.cache_info = cache_info

wrapper.cache_clear = cache_clear

"""Merges MROs in *sequences* to a single MRO using the C3 algorithm.

result = []

while True:

sequences = [s for s in sequences if s] # purge empty sequences

if not sequences:

return result

for s1 in sequences: # find merge candidates among seq heads

candidate = s1[0]

for s2 in sequences:

if candidate in s2[1:]:

candidate = None

break # reject the current head, it appears later

else:

break

if candidate is None:

raise RuntimeError("Inconsistent hierarchy")

result.append(candidate)

# remove the chosen candidate

for seq in sequences:

if seq[0] == candidate:

"""Computes the method resolution order using extended C3 linearization.

for i, base in enumerate(reversed(cls.__bases__)):

if hasattr(base, '__abstractmethods__'):

boundary = len(cls.__bases__) - i

break # Bases up to the last explicit ABC are considered first.

else:

boundary = 0

abcs = list(abcs) if abcs else []

explicit_bases = list(cls.__bases__[:boundary])

abstract_bases = []

other_bases = list(cls.__bases__[boundary:])

for base in abcs:

if issubclass(cls, base) and not any(

issubclass(b, base) for b in cls.__bases__

):

# If *cls* is the class that introduces behaviour described by

# an ABC *base*, insert said ABC to its MRO.

abstract_bases.append(base)

for base in abstract_bases:

abcs.remove(base)

explicit_c3_mros = [_c3_mro(base, abcs=abcs) for base in explicit_bases]

abstract_c3_mros = [_c3_mro(base, abcs=abcs) for base in abstract_bases]

other_c3_mros = [_c3_mro(base, abcs=abcs) for base in other_bases]

return _c3_merge(

[[cls]] +

explicit_c3_mros + abstract_c3_mros + other_c3_mros +

[explicit_bases] + [abstract_bases] + [other_bases]

"""Calculates the method resolution order for a given class *cls*.

bases = set(cls.__mro__)

# Remove entries which are already present in the __mro__ or unrelated.

def is_related(typ):

return (typ not in bases and hasattr(typ, '__mro__')

and not isinstance(typ, GenericAlias)

and issubclass(cls, typ))

types = [n for n in types if is_related(n)]

# Remove entries which are strict bases of other entries (they will end up

# in the MRO anyway.

def is_strict_base(typ):

for other in types:

if typ != other and typ in other.__mro__:

return True

return False

types = [n for n in types if not is_strict_base(n)]

# Subclasses of the ABCs in *types* which are also implemented by

# *cls* can be used to stabilize ABC ordering.

type_set = set(types)

mro = []

for typ in types:

found = []

for sub in typ.__subclasses__():

if sub not in bases and issubclass(cls, sub):

found.append([s for s in sub.__mro__ if s in type_set])

if not found:

mro.append(typ)

continue

# Favor subclasses with the biggest number of useful bases

found.sort(key=len, reverse=True)

for sub in found:

for subcls in sub:

if subcls not in mro:

mro.append(subcls)

"""Returns the best matching implementation from *registry* for type *cls*.

mro = _compose_mro(cls, registry.keys())

match = None

for t in mro:

if match is not None:

# If *match* is an implicit ABC but there is another unrelated,

# equally matching implicit ABC, refuse the temptation to guess.

if (t in registry and t not in cls.__mro__

and match not in cls.__mro__

and not issubclass(match, t)):

raise RuntimeError("Ambiguous dispatch: {} or {}".format(

match, t))

break

if t in registry:

match = t

"""Single-dispatch generic function decorator.

# There are many programs that use functools without singledispatch, so we

# trade-off making singledispatch marginally slower for the benefit of

# making start-up of such applications slightly faster.

import types, weakref

registry = {}

dispatch_cache = weakref.WeakKeyDictionary()

cache_token = None

def dispatch(cls):

"""generic_func.dispatch(cls) -> <function implementation>

nonlocal cache_token

if cache_token is not None:

current_token = get_cache_token()

if cache_token != current_token:

dispatch_cache.clear()

cache_token = current_token

try:

impl = dispatch_cache[cls]

except KeyError:

try:

impl = registry[cls]

except KeyError:

impl = _find_impl(cls, registry)

dispatch_cache[cls] = impl

return impl

def _is_union_type(cls):

from typing import get_origin, Union

return get_origin(cls) in {Union, types.UnionType}

def _is_valid_dispatch_type(cls):

if isinstance(cls, type):

return True

from typing import get_args

return (_is_union_type(cls) and

all(isinstance(arg, type) for arg in get_args(cls)))

def register(cls, func=None):

"""generic_func.register(cls, func) -> func

nonlocal cache_token

if _is_valid_dispatch_type(cls):

if func is None:

return lambda f: register(cls, f)

else:

if func is not None:

raise TypeError(

f"Invalid first argument to `register()`. "

f"{cls!r} is not a class or union type."

)

ann = getattr(cls, '__annotations__', {})

if not ann:

raise TypeError(

f"Invalid first argument to `register()`: {cls!r}. "

f"Use either `@register(some_class)` or plain `@register` "

f"on an annotated function."

)

func = cls

# only import typing if annotation parsing is necessary

from typing import get_type_hints

argname, cls = next(iter(get_type_hints(func).items()))

if not _is_valid_dispatch_type(cls):

if _is_union_type(cls):

raise TypeError(

f"Invalid annotation for {argname!r}. "

f"{cls!r} not all arguments are classes."

)

else:

raise TypeError(

f"Invalid annotation for {argname!r}. "

f"{cls!r} is not a class."

)

if _is_union_type(cls):

from typing import get_args

for arg in get_args(cls):

registry[arg] = func

else:

registry[cls] = func

if cache_token is None and hasattr(cls, '__abstractmethods__'):

cache_token = get_cache_token()

dispatch_cache.clear()

return func

def wrapper(*args, **kw):

if not args:

raise TypeError(f'{funcname} requires at least '

'1 positional argument')

return dispatch(args[0].__class__)(*args, **kw)

funcname = getattr(func, '__name__', 'singledispatch function')

registry[object] = func

wrapper.register = register

wrapper.dispatch = dispatch

wrapper.registry = types.MappingProxyType(registry)

wrapper._clear_cache = dispatch_cache.clear

update_wrapper(wrapper, func)

"""Single-dispatch generic method descriptor.

def __init__(self, func):

if not callable(func) and not hasattr(func, "__get__"):

raise TypeError(f"{func!r} is not callable or a descriptor")

self.dispatcher = singledispatch(func)

self.func = func

def register(self, cls, method=None):

"""generic_method.register(cls, func) -> func

return self.dispatcher.register(cls, func=method)

def __get__(self, obj, cls=None):

def _method(*args, **kwargs):

method = self.dispatcher.dispatch(args[0].__class__)

return method.__get__(obj, cls)(*args, **kwargs)

_method.__isabstractmethod__ = self.__isabstractmethod__

_method.register = self.register

update_wrapper(_method, self.func)

return _method

@property

def __isabstractmethod__(self):

def __init__(self, func):

self.func = func

self.attrname = None

self.__doc__ = func.__doc__

self.lock = RLock()

def __set_name__(self, owner, name):

if self.attrname is None:

self.attrname = name

elif name != self.attrname:

raise TypeError(

"Cannot assign the same cached_property to two different names "

f"({self.attrname!r} and {name!r})."

)

def __get__(self, instance, owner=None):

if instance is None:

return self

if self.attrname is None:

raise TypeError(

"Cannot use cached_property instance without calling __set_name__ on it.")

try:

cache = instance.__dict__

except AttributeError: # not all objects have __dict__ (e.g. class defines slots)

msg = (

f"No '__dict__' attribute on {type(instance).__name__!r} "

f"instance to cache {self.attrname!r} property."

)

raise TypeError(msg) from None

val = cache.get(self.attrname, _NOT_FOUND)

if val is _NOT_FOUND:

with self.lock:

# check if another thread filled cache while we awaited lock

val = cache.get(self.attrname, _NOT_FOUND)

if val is _NOT_FOUND: