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 = self.__lt__(other)

if op_result is NotImplemented:

return NotImplemented

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

op_result = self.__lt__(other)

if op_result is NotImplemented:

return NotImplemented

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

op_result = self.__le__(other)

if op_result is NotImplemented:

return NotImplemented

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

op_result = self.__le__(other)

if op_result is NotImplemented:

return NotImplemented

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

op_result = self.__le__(other)

if op_result is NotImplemented:

return NotImplemented

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

op_result = self.__gt__(other)

if op_result is NotImplemented:

return NotImplemented

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

op_result = self.__gt__(other)

if op_result is NotImplemented:

return NotImplemented

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

op_result = self.__ge__(other)

if op_result is NotImplemented:

return NotImplemented

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

op_result = self.__ge__(other)

if op_result is NotImplemented:

return NotImplemented

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

op_result = self.__ge__(other)

if op_result is NotImplemented:

return NotImplemented

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

convert = {

'__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)]

}

# 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

def __ne__(self, other):

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

__hash__ = None

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

def newfunc(*fargs, **fkeywords):

newkeywords = keywords.copy()

newkeywords.update(fkeywords)

return func(*(args + fargs), **newkeywords)

newfunc.func = func

newfunc.args = args

newfunc.keywords = keywords

"""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.copy()

self.keywords.update(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__.__name__,

func=self.func,

args=args,

keywords=keywords)

def _make_unbound_method(self):

def _method(*args, **keywords):

call_keywords = self.keywords.copy()

call_keywords.update(keywords)

cls_or_self, *rest = args

call_args = (cls_or_self,) + self.args + tuple(rest)

return self.func(*call_args, **call_keywords)

_method.__isabstractmethod__ = self.__isabstractmethod__

_method._partialmethod = self

return _method

def __get__(self, obj, cls):

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):

""" 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, frozenset, type(None)},

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

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

key = args

if kwds:

sorted_items = sorted(kwds.items())

key += kwd_mark

for item in sorted_items:

key += item

if typed:

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

if kwds:

key += tuple(type(v) for k, v in sorted_items)

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).

# Early detection of an erroneous call to @lru_cache without any arguments

# resulting in the inner function being passed to maxsize instead of an

# integer or None.

if maxsize is not None and not isinstance(maxsize, int):

raise TypeError('Expected maxsize to be an integer or None')

# 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

def decorating_function(user_function):

cache = {}

hits = misses = 0

full = False

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

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 after a successful call

nonlocal misses

result = user_function(*args, **kwds)

misses += 1

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

result = user_function(*args, **kwds)

cache[key] = result

misses += 1

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

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

full = (len(cache) >= maxsize)

misses += 1

return result

def cache_info():

"""Report cache statistics"""

with lock:

return _CacheInfo(hits, misses, maxsize, len(cache))

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

return update_wrapper(wrapper, user_function)

"""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 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.

registry = {}

dispatch_cache = 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 register(cls, func=None):

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

nonlocal cache_token

if func is None:

return lambda f: register(cls, f)

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):

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

registry[object] = func

wrapper.register = register

wrapper.dispatch = dispatch

wrapper.registry = MappingProxyType(registry)

wrapper._clear_cache = dispatch_cache.clear

update_wrapper(wrapper, func)