venv/lib/python3.11/site-packages/setuptools/_vendor/jaraco/functools.py

   1 import functools
   2 import time
   3 import inspect
   4 import collections
   5 import types
   6 import itertools
   7 import warnings
   8
   9 import setuptools.extern.more_itertools
  10
  11 from typing import Callable, TypeVar
  12
  13
  14 CallableT = TypeVar("CallableT", bound=Callable[..., object])
  15
  16
  17 def compose(*funcs):
  18     """
  19     Compose any number of unary functions into a single unary function.
  20
  21     >>> import textwrap
  22     >>> expected = str.strip(textwrap.dedent(compose.__doc__))
  23     >>> strip_and_dedent = compose(str.strip, textwrap.dedent)
  24     >>> strip_and_dedent(compose.__doc__) == expected
  25     True
  26
  27     Compose also allows the innermost function to take arbitrary arguments.
  28
  29     >>> round_three = lambda x: round(x, ndigits=3)
  30     >>> f = compose(round_three, int.__truediv__)
  31     >>> [f(3*x, x+1) for x in range(1,10)]
  32     [1.5, 2.0, 2.25, 2.4, 2.5, 2.571, 2.625, 2.667, 2.7]
  33     """
  34
  35     def compose_two(f1, f2):
  36         return lambda *args, **kwargs: f1(f2(*args, **kwargs))
  37
  38     return functools.reduce(compose_two, funcs)
  39
  40
  41 def method_caller(method_name, *args, **kwargs):
  42     """
  43     Return a function that will call a named method on the
  44     target object with optional positional and keyword
  45     arguments.
  46
  47     >>> lower = method_caller('lower')
  48     >>> lower('MyString')
  49     'mystring'
  50     """
  51
  52     def call_method(target):
  53         func = getattr(target, method_name)
  54         return func(*args, **kwargs)
  55
  56     return call_method
  57
  58
  59 def once(func):
  60     """
  61     Decorate func so it's only ever called the first time.
  62
  63     This decorator can ensure that an expensive or non-idempotent function
  64     will not be expensive on subsequent calls and is idempotent.
  65
  66     >>> add_three = once(lambda a: a+3)
  67     >>> add_three(3)
  68     6
  69     >>> add_three(9)
  70     6
  71     >>> add_three('12')
  72     6
  73
  74     To reset the stored value, simply clear the property ``saved_result``.
  75
  76     >>> del add_three.saved_result
  77     >>> add_three(9)
  78     12
  79     >>> add_three(8)
  80     12
  81
  82     Or invoke 'reset()' on it.
  83
  84     >>> add_three.reset()
  85     >>> add_three(-3)
  86     0
  87     >>> add_three(0)
  88     0
  89     """
  90
  91     @functools.wraps(func)
  92     def wrapper(*args, **kwargs):
  93         if not hasattr(wrapper, 'saved_result'):
  94             wrapper.saved_result = func(*args, **kwargs)
  95         return wrapper.saved_result
  96
  97     wrapper.reset = lambda: vars(wrapper).__delitem__('saved_result')
  98     return wrapper
  99
 100
 101 def method_cache(
 102     method: CallableT,
 103     cache_wrapper: Callable[
 104         [CallableT], CallableT
 105     ] = functools.lru_cache(),  # type: ignore[assignment]
 106 ) -> CallableT:
 107     """
 108     Wrap lru_cache to support storing the cache data in the object instances.
 109
 110     Abstracts the common paradigm where the method explicitly saves an
 111     underscore-prefixed protected property on first call and returns that
 112     subsequently.
 113
 114     >>> class MyClass:
 115     ...     calls = 0
 116     ...
 117     ...     @method_cache
 118     ...     def method(self, value):
 119     ...         self.calls += 1
 120     ...         return value
 121
 122     >>> a = MyClass()
 123     >>> a.method(3)
 124     3
 125     >>> for x in range(75):
 126     ...     res = a.method(x)
 127     >>> a.calls
 128     75
 129
 130     Note that the apparent behavior will be exactly like that of lru_cache
 131     except that the cache is stored on each instance, so values in one
 132     instance will not flush values from another, and when an instance is
 133     deleted, so are the cached values for that instance.
 134
 135     >>> b = MyClass()
 136     >>> for x in range(35):
 137     ...     res = b.method(x)
 138     >>> b.calls
 139     35
 140     >>> a.method(0)
 141     0
 142     >>> a.calls
 143     75
 144
 145     Note that if method had been decorated with ``functools.lru_cache()``,
 146     a.calls would have been 76 (due to the cached value of 0 having been
 147     flushed by the 'b' instance).
 148
 149     Clear the cache with ``.cache_clear()``
 150
 151     >>> a.method.cache_clear()
 152
 153     Same for a method that hasn't yet been called.
 154
 155     >>> c = MyClass()
 156     >>> c.method.cache_clear()
 157
 158     Another cache wrapper may be supplied:
 159
 160     >>> cache = functools.lru_cache(maxsize=2)
 161     >>> MyClass.method2 = method_cache(lambda self: 3, cache_wrapper=cache)
 162     >>> a = MyClass()
 163     >>> a.method2()
 164     3
 165
 166     Caution - do not subsequently wrap the method with another decorator, such
 167     as ``@property``, which changes the semantics of the function.
 168
 169     See also
 170     http://code.activestate.com/recipes/577452-a-memoize-decorator-for-instance-methods/
 171     for another implementation and additional justification.
 172     """
 173
 174     def wrapper(self: object, *args: object, **kwargs: object) -> object:
 175         # it's the first call, replace the method with a cached, bound method
 176         bound_method: CallableT = types.MethodType(  # type: ignore[assignment]
 177             method, self
 178         )
 179         cached_method = cache_wrapper(bound_method)
 180         setattr(self, method.__name__, cached_method)
 181         return cached_method(*args, **kwargs)
 182
 183     # Support cache clear even before cache has been created.
 184     wrapper.cache_clear = lambda: None  # type: ignore[attr-defined]
 185
 186     return (  # type: ignore[return-value]
 187         _special_method_cache(method, cache_wrapper) or wrapper
 188     )
 189
 190
 191 def _special_method_cache(method, cache_wrapper):
 192     """
 193     Because Python treats special methods differently, it's not
 194     possible to use instance attributes to implement the cached
 195     methods.
 196
 197     Instead, install the wrapper method under a different name
 198     and return a simple proxy to that wrapper.
 199
 200     https://github.com/jaraco/jaraco.functools/issues/5
 201     """
 202     name = method.__name__
 203     special_names = '__getattr__', '__getitem__'
 204     if name not in special_names:
 205         return
 206
 207     wrapper_name = '__cached' + name
 208
 209     def proxy(self, *args, **kwargs):
 210         if wrapper_name not in vars(self):
 211             bound = types.MethodType(method, self)
 212             cache = cache_wrapper(bound)
 213             setattr(self, wrapper_name, cache)
 214         else:
 215             cache = getattr(self, wrapper_name)
 216         return cache(*args, **kwargs)
 217
 218     return proxy
 219
 220
 221 def apply(transform):
 222     """
 223     Decorate a function with a transform function that is
 224     invoked on results returned from the decorated function.
 225
 226     >>> @apply(reversed)
 227     ... def get_numbers(start):
 228     ...     "doc for get_numbers"
 229     ...     return range(start, start+3)
 230     >>> list(get_numbers(4))
 231     [6, 5, 4]
 232     >>> get_numbers.__doc__
 233     'doc for get_numbers'
 234     """
 235
 236     def wrap(func):
 237         return functools.wraps(func)(compose(transform, func))
 238
 239     return wrap
 240
 241
 242 def result_invoke(action):
 243     r"""
 244     Decorate a function with an action function that is
 245     invoked on the results returned from the decorated
 246     function (for its side-effect), then return the original
 247     result.
 248
 249     >>> @result_invoke(print)
 250     ... def add_two(a, b):
 251     ...     return a + b
 252     >>> x = add_two(2, 3)
 253     5
 254     >>> x
 255     5
 256     """
 257
 258     def wrap(func):
 259         @functools.wraps(func)
 260         def wrapper(*args, **kwargs):
 261             result = func(*args, **kwargs)
 262             action(result)
 263             return result
 264
 265         return wrapper
 266
 267     return wrap
 268
 269
 270 def invoke(f, *args, **kwargs):
 271     """
 272     Call a function for its side effect after initialization.
 273
 274     The benefit of using the decorator instead of simply invoking a function
 275     after defining it is that it makes explicit the author's intent for the
 276     function to be called immediately. Whereas if one simply calls the
 277     function immediately, it's less obvious if that was intentional or
 278     incidental. It also avoids repeating the name - the two actions, defining
 279     the function and calling it immediately are modeled separately, but linked
 280     by the decorator construct.
 281
 282     The benefit of having a function construct (opposed to just invoking some
 283     behavior inline) is to serve as a scope in which the behavior occurs. It
 284     avoids polluting the global namespace with local variables, provides an
 285     anchor on which to attach documentation (docstring), keeps the behavior
 286     logically separated (instead of conceptually separated or not separated at
 287     all), and provides potential to re-use the behavior for testing or other
 288     purposes.
 289
 290     This function is named as a pithy way to communicate, "call this function
 291     primarily for its side effect", or "while defining this function, also
 292     take it aside and call it". It exists because there's no Python construct
 293     for "define and call" (nor should there be, as decorators serve this need
 294     just fine). The behavior happens immediately and synchronously.
 295
 296     >>> @invoke
 297     ... def func(): print("called")
 298     called
 299     >>> func()
 300     called
 301
 302     Use functools.partial to pass parameters to the initial call
 303
 304     >>> @functools.partial(invoke, name='bingo')
 305     ... def func(name): print("called with", name)
 306     called with bingo
 307     """
 308     f(*args, **kwargs)
 309     return f
 310
 311
 312 def call_aside(*args, **kwargs):
 313     """
 314     Deprecated name for invoke.
 315     """
 316     warnings.warn("call_aside is deprecated, use invoke", DeprecationWarning)
 317     return invoke(*args, **kwargs)
 318
 319
 320 class Throttler:
 321     """
 322     Rate-limit a function (or other callable)
 323     """
 324
 325     def __init__(self, func, max_rate=float('Inf')):
 326         if isinstance(func, Throttler):
 327             func = func.func
 328         self.func = func
 329         self.max_rate = max_rate
 330         self.reset()
 331
 332     def reset(self):
 333         self.last_called = 0
 334
 335     def __call__(self, *args, **kwargs):
 336         self._wait()
 337         return self.func(*args, **kwargs)
 338
 339     def _wait(self):
 340         "ensure at least 1/max_rate seconds from last call"
 341         elapsed = time.time() - self.last_called
 342         must_wait = 1 / self.max_rate - elapsed
 343         time.sleep(max(0, must_wait))
 344         self.last_called = time.time()
 345
 346     def __get__(self, obj, type=None):
 347         return first_invoke(self._wait, functools.partial(self.func, obj))
 348
 349
 350 def first_invoke(func1, func2):
 351     """
 352     Return a function that when invoked will invoke func1 without
 353     any parameters (for its side-effect) and then invoke func2
 354     with whatever parameters were passed, returning its result.
 355     """
 356
 357     def wrapper(*args, **kwargs):
 358         func1()
 359         return func2(*args, **kwargs)
 360
 361     return wrapper
 362
 363
 364 def retry_call(func, cleanup=lambda: None, retries=0, trap=()):
 365     """
 366     Given a callable func, trap the indicated exceptions
 367     for up to 'retries' times, invoking cleanup on the
 368     exception. On the final attempt, allow any exceptions
 369     to propagate.
 370     """
 371     attempts = itertools.count() if retries == float('inf') else range(retries)
 372     for attempt in attempts:
 373         try:
 374             return func()
 375         except trap:
 376             cleanup()
 377
 378     return func()
 379
 380
 381 def retry(*r_args, **r_kwargs):
 382     """
 383     Decorator wrapper for retry_call. Accepts arguments to retry_call
 384     except func and then returns a decorator for the decorated function.
 385
 386     Ex:
 387
 388     >>> @retry(retries=3)
 389     ... def my_func(a, b):
 390     ...     "this is my funk"
 391     ...     print(a, b)
 392     >>> my_func.__doc__
 393     'this is my funk'
 394     """
 395
 396     def decorate(func):
 397         @functools.wraps(func)
 398         def wrapper(*f_args, **f_kwargs):
 399             bound = functools.partial(func, *f_args, **f_kwargs)
 400             return retry_call(bound, *r_args, **r_kwargs)
 401
 402         return wrapper
 403
 404     return decorate
 405
 406
 407 def print_yielded(func):
 408     """
 409     Convert a generator into a function that prints all yielded elements
 410
 411     >>> @print_yielded
 412     ... def x():
 413     ...     yield 3; yield None
 414     >>> x()
 415     3
 416     None
 417     """
 418     print_all = functools.partial(map, print)
 419     print_results = compose(more_itertools.consume, print_all, func)
 420     return functools.wraps(func)(print_results)
 421
 422
 423 def pass_none(func):
 424     """
 425     Wrap func so it's not called if its first param is None
 426
 427     >>> print_text = pass_none(print)
 428     >>> print_text('text')
 429     text
 430     >>> print_text(None)
 431     """
 432
 433     @functools.wraps(func)
 434     def wrapper(param, *args, **kwargs):
 435         if param is not None:
 436             return func(param, *args, **kwargs)
 437
 438     return wrapper
 439
 440
 441 def assign_params(func, namespace):
 442     """
 443     Assign parameters from namespace where func solicits.
 444
 445     >>> def func(x, y=3):
 446     ...     print(x, y)
 447     >>> assigned = assign_params(func, dict(x=2, z=4))
 448     >>> assigned()
 449     2 3
 450
 451     The usual errors are raised if a function doesn't receive
 452     its required parameters:
 453
 454     >>> assigned = assign_params(func, dict(y=3, z=4))
 455     >>> assigned()
 456     Traceback (most recent call last):
 457     TypeError: func() ...argument...
 458
 459     It even works on methods:
 460
 461     >>> class Handler:
 462     ...     def meth(self, arg):
 463     ...         print(arg)
 464     >>> assign_params(Handler().meth, dict(arg='crystal', foo='clear'))()
 465     crystal
 466     """
 467     sig = inspect.signature(func)
 468     params = sig.parameters.keys()
 469     call_ns = {k: namespace[k] for k in params if k in namespace}
 470     return functools.partial(func, **call_ns)
 471
 472
 473 def save_method_args(method):
 474     """
 475     Wrap a method such that when it is called, the args and kwargs are
 476     saved on the method.
 477
 478     >>> class MyClass:
 479     ...     @save_method_args
 480     ...     def method(self, a, b):
 481     ...         print(a, b)
 482     >>> my_ob = MyClass()
 483     >>> my_ob.method(1, 2)
 484     1 2
 485     >>> my_ob._saved_method.args
 486     (1, 2)
 487     >>> my_ob._saved_method.kwargs
 488     {}
 489     >>> my_ob.method(a=3, b='foo')
 490     3 foo
 491     >>> my_ob._saved_method.args
 492     ()
 493     >>> my_ob._saved_method.kwargs == dict(a=3, b='foo')
 494     True
 495
 496     The arguments are stored on the instance, allowing for
 497     different instance to save different args.
 498
 499     >>> your_ob = MyClass()
 500     >>> your_ob.method({str('x'): 3}, b=[4])
 501     {'x': 3} [4]
 502     >>> your_ob._saved_method.args
 503     ({'x': 3},)
 504     >>> my_ob._saved_method.args
 505     ()
 506     """
 507     args_and_kwargs = collections.namedtuple('args_and_kwargs', 'args kwargs')
 508
 509     @functools.wraps(method)
 510     def wrapper(self, *args, **kwargs):
 511         attr_name = '_saved_' + method.__name__
 512         attr = args_and_kwargs(args, kwargs)
 513         setattr(self, attr_name, attr)
 514         return method(self, *args, **kwargs)
 515
 516     return wrapper
 517
 518
 519 def except_(*exceptions, replace=None, use=None):
 520     """
 521     Replace the indicated exceptions, if raised, with the indicated
 522     literal replacement or evaluated expression (if present).
 523
 524     >>> safe_int = except_(ValueError)(int)
 525     >>> safe_int('five')
 526     >>> safe_int('5')
 527     5
 528
 529     Specify a literal replacement with ``replace``.
 530
 531     >>> safe_int_r = except_(ValueError, replace=0)(int)
 532     >>> safe_int_r('five')
 533     0
 534
 535     Provide an expression to ``use`` to pass through particular parameters.
 536
 537     >>> safe_int_pt = except_(ValueError, use='args[0]')(int)
 538     >>> safe_int_pt('five')
 539     'five'
 540
 541     """
 542
 543     def decorate(func):
 544         @functools.wraps(func)
 545         def wrapper(*args, **kwargs):
 546             try:
 547                 return func(*args, **kwargs)
 548             except exceptions:
 549                 try:
 550                     return eval(use)
 551                 except TypeError:
 552                     return replace
 553
 554         return wrapper
 555
 556     return decorate