Maintainers
Pricing
Enterprise
More
More
Documentation
Try Live
Rankings
Install
Blog
Docs7
Ask
New
Add Docs
propcache
https://propcache.aio-libs.org/en/latest/
Admin
propcache is a Python package that provides accelerated and custom implementations of
...
Tokens:
11,542
Snippets:
75
Trust Score:
6.4
Update:
2 months ago
Context
Chat
Benchmark
Latest
Show doc for...
Code
Info
Show Results
Tokens:
Raw
Copy
Link
Tokens:
Raw
Copy
Link
### Changelog fragment examples Source: https://propcache.aio-libs.org/en/latest/contributing/guidelines Examples of Towncrier changelog files formatted in reStructuredText, placed in the CHANGES/ directory. ```rst Dropped Python 3.5 support; Python 3.6 is the minimal supported Python version -- by :user:`webknjaz`. ``` ```rst Started shipping Windows wheels for the x86 architecture -- by :user:`Dreamsorcerer`. ``` ```rst Added support for ``GenericAliases`` (``MultiDict[str]``) under Python 3.9 and higher -- by :user:`mjpieters`. ``` -------------------------------- ### Build documentation locally Source: https://propcache.aio-libs.org/en/latest/_sources/contributing/guidelines.rst.txt Execute this command in the terminal to trigger the local documentation build process. ```shell-session $ make doc ``` -------------------------------- ### Prepare Version and Changelog Source: https://propcache.aio-libs.org/en/latest/_sources/contributing/release_guide.rst.txt Update the version string and generate the changelog using towncrier before committing the changes. ```shell-session [dir:propcache] $ propcache/__init__.py [dir:propcache] $ python -m towncrier build \ -- --version 'VERSION_WITHOUT_LEADING_V' [dir:propcache] $ git commit -v CHANGES{.rst,/} propcache/__init__.py ``` -------------------------------- ### Tag the Release Source: https://propcache.aio-libs.org/en/latest/_sources/contributing/release_guide.rst.txt Create a GPG-signed Git tag for the release with a descriptive message. ```shell-session [dir:propcache] $ git tag \ -s 'VERSION_WITH_LEADING_V' \ -m 'VERSION_WITH_LEADING_V' \ -m 'This release does X and Y.' ``` -------------------------------- ### Configure build-inplace setting Source: https://propcache.aio-libs.org/en/latest/changes Use this command to control whether the project is built in-tree or in a temporary directory during the build process. ```bash $ python -m build --config-setting=build-inplace=true ``` -------------------------------- ### Importing C-based LRU Cache Source: https://propcache.aio-libs.org/en/latest/_modules/functools Attempt to import the optimized C implementation of the LRU cache wrapper. ```python try: from _functools import _lru_cache_wrapper except ImportError: pass ``` -------------------------------- ### Push Release Tag Source: https://propcache.aio-libs.org/en/latest/_sources/contributing/release_guide.rst.txt Push the newly created version tag to the upstream repository. ```shell-session [dir:propcache] $ git push origin 'VERSION_WITH_LEADING_V' ``` -------------------------------- ### Find implementation in registry Source: https://propcache.aio-libs.org/en/latest/_modules/functools Retrieves the best matching implementation from a registry for a given class type. ```python def _find_impl(cls, registry): """Returns the best matching implementation from *registry* for type *cls*. Where there is no registered implementation for a specific type, its method resolution order is used to find a more generic implementation. Note: if *registry* does not contain an implementation for the base *object* type, this function may return None. """ 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 return registry.get(match) ``` -------------------------------- ### Partial Method Implementation Source: https://propcache.aio-libs.org/en/latest/_modules/functools Internal methods for handling partial function binding and descriptor protocol integration. ```python 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) __class_getitem__ = classmethod(GenericAlias) ``` -------------------------------- ### Comparison to key converter Source: https://propcache.aio-libs.org/en/latest/_modules/functools Converts a legacy cmp function into a key function for sorting. ```python def cmp_to_key(mycmp): """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 return K try: from _functools import cmp_to_key except ImportError: pass ``` -------------------------------- ### Using under_cached_property Source: https://propcache.aio-libs.org/en/latest/_sources/api.rst.txt Demonstrates defining a class with a cached property that stores results in a private _cache dictionary. ```python from propcache.api import under_cached_property class MyClass: def __init__(self, data: List[float]): self._data = data self._cache = {} @under_cached_property def calculated_data(self): return expensive_operation(self._data) def clear_cache(self): self._cache.clear() instance = MyClass([1.0, 2.0, 3.0]) print(instance.calculated_data) # expensive operation instance.clear_cache() print(instance.calculated_data) # expensive operation ``` -------------------------------- ### Implement reduce logic Source: https://propcache.aio-libs.org/en/latest/_modules/functools Core logic for the reduce function, handling empty iterables and initial values. ```python 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) return value try: from _functools import reduce except ImportError: pass ``` -------------------------------- ### Compute MRO with extended C3 linearization Source: https://propcache.aio-libs.org/en/latest/_modules/functools Calculates the method resolution order for a class, optionally inserting abstract base classes. ```python def _c3_mro(cls, abcs=None): """Computes the method resolution order using extended C3 linearization. If no *abcs* are given, the algorithm works exactly like the built-in C3 linearization used for method resolution. If given, *abcs* is a list of abstract base classes that should be inserted into the resulting MRO. Unrelated ABCs are ignored and don't end up in the result. The algorithm inserts ABCs where their functionality is introduced, i.e. issubclass(cls, abc) returns True for the class itself but returns False for all its direct base classes. Implicit ABCs for a given class (either registered or inferred from the presence of a special method like __len__) are inserted directly after the last ABC explicitly listed in the MRO of said class. If two implicit ABCs end up next to each other in the resulting MRO, their ordering depends on the order of types in *abcs*. """ 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] ) ``` -------------------------------- ### LRU Cache Decorator Logic Source: https://propcache.aio-libs.org/en/latest/_modules/functools Handles the initialization of the LRU cache decorator, including argument validation and the creation of the decorating function. ```python 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) return decorating_function ``` -------------------------------- ### Ordering method conversion helpers Source: https://propcache.aio-libs.org/en/latest/_modules/functools Helper functions used by total_ordering to derive missing comparison methods from existing ones. ```python def _lt_from_le(self, other): '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 op_result and self != other def _gt_from_le(self, other): '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 not op_result def _lt_from_gt(self, other): '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 not op_result and self != other def _ge_from_gt(self, other): '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 op_result or self == other def _le_from_gt(self, other): '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 not op_result def _le_from_ge(self, other): '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 not op_result or self == other def _gt_from_ge(self, other): '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 op_result and self != other def _lt_from_ge(self, other): '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 return not op_result ``` -------------------------------- ### Single Dispatch Method Descriptor Source: https://propcache.aio-libs.org/en/latest/_modules/functools Implementation of the singledispatchmethod descriptor for class-based generic dispatching. ```python class singledispatchmethod: """Single-dispatch generic method descriptor. Supports wrapping existing descriptors and handles non-descriptor callables as instance methods. """ 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 Registers a new implementation for the given *cls* on a *generic_method*. """ return self.dispatcher.register(cls, func=method) ``` -------------------------------- ### Generic Function Dispatch Implementation Source: https://propcache.aio-libs.org/en/latest/_modules/functools Internal logic for managing function registries and dispatching based on the type of the first argument. ```python import types, weakref registry = {} dispatch_cache = weakref.WeakKeyDictionary() cache_token = None def dispatch(cls): """generic_func.dispatch(cls) -> <function implementation> Runs the dispatch algorithm to return the best available implementation for the given *cls* registered on *generic_func*. """ 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 Registers a new implementation for the given *cls* on a *generic_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) return wrapper ``` -------------------------------- ### Define partialmethod class Source: https://propcache.aio-libs.org/en/latest/_modules/functools A method descriptor class for partial application, handling descriptors and non-descriptor callables. ```python class partialmethod(object): """Method descriptor with partial application of the given arguments and keywords. Supports wrapping existing descriptors and handles non-descriptor callables as instance methods. """ 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) ``` -------------------------------- ### LRU Cache Wrapper Implementation Source: https://propcache.aio-libs.org/en/latest/_modules/functools Internal wrapper function for size-limited caching that tracks access recency using a circular queue. ```python 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 return wrapper ``` -------------------------------- ### C3 MRO Merge Algorithm Source: https://propcache.aio-libs.org/en/latest/_modules/functools Merges multiple Method Resolution Orders into a single MRO using the C3 linearization algorithm. ```python def _c3_merge(sequences): """Merges MROs in *sequences* to a single MRO using the C3 algorithm. Adapted from https://www.python.org/download/releases/2.3/mro/. """ 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: del seq[0] ``` -------------------------------- ### Total ordering helper functions Source: https://propcache.aio-libs.org/en/latest/_modules/functools Internal helper functions used by the total_ordering decorator to derive comparison operators from existing ones. ```python def _gt_from_lt(self, other): '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 not op_result and self != other def _le_from_lt(self, other): '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 op_result or self == other def _ge_from_lt(self, other): '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 not op_result def _ge_from_le(self, other): '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 not op_result or self == other ``` -------------------------------- ### Define partial class Source: https://propcache.aio-libs.org/en/latest/_modules/functools A class for partial function application, supporting slots and custom state reduction. ```python class partial: """New function with partial application of the given arguments and keywords. """ __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 self.keywords = kwds try: from _functools import partial except ImportError: pass ``` -------------------------------- ### Partial Unwrapping Helper Source: https://propcache.aio-libs.org/en/latest/_modules/functools Utility function to recursively unwrap nested partial objects. ```python def _unwrap_partial(func): while isinstance(func, partial): func = func.func return func ``` -------------------------------- ### LRU Cache Wrapper Implementation Source: https://propcache.aio-libs.org/en/latest/_modules/functools Defines the internal wrapper function that manages cache state, including the linked list structure for tracking cache entries. ```python def _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo): # 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: ``` -------------------------------- ### Wraps decorator factory Source: https://propcache.aio-libs.org/en/latest/_modules/functools A convenience decorator factory that applies update_wrapper to a wrapper function. ```python def wraps(wrapped, assigned = WRAPPER_ASSIGNMENTS, updated = WRAPPER_UPDATES): """Decorator factory to apply update_wrapper() to a wrapper function Returns a decorator that invokes update_wrapper() with the decorated function as the wrapper argument and the arguments to wraps() as the remaining arguments. Default arguments are as for update_wrapper(). This is a convenience function to simplify applying partial() to update_wrapper(). """ return partial(update_wrapper, wrapped=wrapped, assigned=assigned, updated=updated) ``` -------------------------------- ### Single-dispatch decorator Source: https://propcache.aio-libs.org/en/latest/_modules/functools Placeholder for the single-dispatch generic function decorator. ```python def singledispatch(func): """Single-dispatch generic function decorator. ``` -------------------------------- ### LRU Cache Internal Structures Source: https://propcache.aio-libs.org/en/latest/_modules/functools Supporting classes and key generation logic for the lru_cache decorator. ```python _CacheInfo = namedtuple("CacheInfo", ["hits", "misses", "maxsize", "currsize"]) class _HashedSeq(list): """ This class guarantees that hash() will be called no more than once per element. This is important because the lru_cache() will hash the key multiple times on a cache miss. """ __slots__ = 'hashvalue' def __init__(self, tup, hash=hash): self[:] = tup self.hashvalue = hash(tup) def __hash__(self): return self.hashvalue def _make_key(args, kwds, typed, kwd_mark = (object(),), fasttypes = {int, str}, tuple=tuple, type=type, len=len): """Make a cache key from optionally typed positional and keyword arguments The key is constructed in a way that is flat as possible rather than as a nested structure that would take more memory. If there is only a single argument and its data type is known to cache its hash value, then that argument is returned without a wrapper. This saves space and improves lookup speed. """ # 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] return _HashedSeq(key) ``` -------------------------------- ### Unbounded Cache Decorator Source: https://propcache.aio-libs.org/en/latest/_modules/functools A lightweight decorator providing an unbounded cache, often referred to as memoization. ```python def cache(user_function, /): 'Simple lightweight unbounded cache. Sometimes called "memoize".' return lru_cache(maxsize=None)(user_function) ``` -------------------------------- ### Compose MRO for a class Source: https://propcache.aio-libs.org/en/latest/_modules/functools Calculates the MRO for a class including relevant abstract base classes from a provided iterable. ```python def _compose_mro(cls, types): """Calculates the method resolution order for a given class *cls*. Includes relevant abstract base classes (with their respective bases) from the *types* iterable. Uses a modified C3 linearization algorithm. """ 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) return _c3_mro(cls, abcs=mro) ``` -------------------------------- ### Update wrapper function metadata Source: https://propcache.aio-libs.org/en/latest/_modules/functools Updates a wrapper function to inherit attributes from the wrapped function, ensuring proper introspection. ```python WRAPPER_ASSIGNMENTS = ('__module__', '__name__', '__qualname__', '__doc__', '__annotations__') WRAPPER_UPDATES = ('__dict__',) def update_wrapper(wrapper, wrapped, assigned = WRAPPER_ASSIGNMENTS, updated = WRAPPER_UPDATES): """Update a wrapper function to look like the wrapped function wrapper is the function to be updated wrapped is the original function assigned is a tuple naming the attributes assigned directly from the wrapped function to the wrapper function (defaults to functools.WRAPPER_ASSIGNMENTS) updated is a tuple naming the attributes of the wrapper that are updated with the corresponding attribute from the wrapped function (defaults to functools.WRAPPER_UPDATES) """ 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() return wrapper ``` -------------------------------- ### Total ordering decorator Source: https://propcache.aio-libs.org/en/latest/_modules/functools Class decorator that fills in missing ordering methods based on defined comparison operations. ```python _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)] } def total_ordering(cls): """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) return cls ``` -------------------------------- ### Implement cached_property class Source: https://propcache.aio-libs.org/en/latest/_modules/functools Defines the cached_property class for computing and caching properties on an instance. ```python class cached_property: 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: val = self.func(instance) try: cache[self.attrname] = val except TypeError: msg = ( f"The '__dict__' attribute on {type(instance).__name__!r} instance " f"does not support item assignment for caching {self.attrname!r} property." ) raise TypeError(msg) from None return val __class_getitem__ = classmethod(GenericAlias) ``` -------------------------------- ### Define under_cached_property descriptor Source: https://propcache.aio-libs.org/en/latest/_modules/propcache/_helpers_py The under_cached_property class acts as a data descriptor that caches the result of a method call in the instance's _cache dictionary. ```python """Various helper functions.""" from __future__ import annotations import sys from collections.abc import Callable, Mapping from functools import cached_property from typing import Any, Generic, Protocol, TypeVar, overload __all__ = ("under_cached_property", "cached_property") if sys.version_info >= (3, 11): from typing import Self else: Self = Any _T = TypeVar("_T") # We use Mapping to make it possible to use TypedDict, but this isn't # technically type safe as we need to assign into the dict. _Cache = TypeVar("_Cache", bound=Mapping[str, Any]) class _CacheImpl(Protocol[_Cache]): _cache: _Cache [docs] class under_cached_property(Generic[_T]): """Use as a class method decorator. It operates almost exactly like the Python `@property` decorator, but it puts the result of the method it decorates into the instance dict after the first call, effectively replacing the function it decorates with an instance variable. It is, in Python parlance, a data descriptor. """ def __init__(self, wrapped: Callable[[Any], _T]) -> None: self.wrapped = wrapped self.__doc__ = wrapped.__doc__ self.name = wrapped.__name__ @overload def __get__(self, inst: None, owner: type[object] | None = None) -> Self: ... @overload def __get__( self, inst: _CacheImpl[Any], owner: type[object] | None = None ) -> _T: ... def __get__( self, inst: _CacheImpl[Any] | None, owner: type[object] | None = None ) -> _T | Self: if inst is None: return self try: return inst._cache[self.name] # type: ignore[no-any-return] except KeyError: val = self.wrapped(inst) inst._cache[self.name] = val return val def __set__(self, inst: _CacheImpl[Any], value: _T) -> None: raise AttributeError("cached property is read-only") ``` -------------------------------- ### LRU Cache Decorator Definition Source: https://propcache.aio-libs.org/en/latest/_modules/functools Public interface for the least-recently-used cache decorator. ```python def lru_cache(maxsize=128, typed=False): """Least-recently-used cache decorator. If *maxsize* is set to None, the LRU features are disabled and the cache can grow without bound. If *typed* is True, arguments of different types will be cached separately. For example, f(3.0) and f(3) will be treated as distinct calls with distinct results. Arguments to the cached function must be hashable. View the cache statistics named tuple (hits, misses, maxsize, currsize) with f.cache_info(). Clear the cache and statistics with f.cache_clear(). Access the underlying function with f.__wrapped__. See: https://en.wikipedia.org/wiki/Cache_replacement_policies#Least_recently_used_(LRU) """ # 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). ``` -------------------------------- ### Reduce sequence function Source: https://propcache.aio-libs.org/en/latest/_modules/functools Applies a function of two arguments cumulatively to items of an iterable. ```python _initial_missing = object() def reduce(function, sequence, initial=_initial_missing): """ reduce(function, iterable[, initial]) -> value Apply a function of two arguments cumulatively to the items of a sequence or iterable, from left to right, so as to reduce the iterable to a single value. For example, reduce(lambda x, y: x+y, [1, 2, 3, 4, 5]) calculates ((((1+2)+3)+4)+5). If initial is present, it is placed before the items of the iterable in the calculation, and serves as a default when the iterable is empty. """ it = iter(sequence) ``` -------------------------------- ### under_cached_property(func) Source: https://propcache.aio-libs.org/en/latest/_sources/api.rst.txt A decorator that caches the result of a method call in the instance's _cache dictionary as a private attribute. ```APIDOC ## under_cached_property(func) ### Description Transforms a method into a property whose value is computed once and cached in the instance's _cache dictionary. This is useful for keeping the main __dict__ clean. ### Parameters - **func** (callable) - The method to be decorated. ### Usage ```python from propcache.api import under_cached_property class MyClass: def __init__(self): self._cache = {} @under_cached_property def calculated_data(self): return expensive_operation() ``` ``` -------------------------------- ### under_cached_property Source: https://propcache.aio-libs.org/en/latest/_modules/propcache/_helpers_py A class method decorator that acts as a data descriptor. It caches the result of the decorated method in the instance dictionary after the first call, effectively replacing the function with an instance variable. ```APIDOC ## class under_cached_property(Generic[_T]) ### Description Use as a class method decorator. It operates almost exactly like the Python `@property` decorator, but it puts the result of the method it decorates into the instance dict after the first call, effectively replacing the function it decorates with an instance variable. ### Usage Decorate a method within a class that implements a `_cache` mapping (e.g., `_CacheImpl`). ### Parameters - **wrapped** (Callable[[Any], _T]) - Required - The method to be decorated and cached. ### Behavior - **__get__**: Returns the cached value from `inst._cache` if it exists; otherwise, calls the wrapped method, stores the result in `inst._cache`, and returns it. - **__set__**: Raises an `AttributeError` as the property is read-only. ``` -------------------------------- ### cached_property(func) Source: https://propcache.aio-libs.org/en/latest/_sources/api.rst.txt A decorator that caches the result of a method call in the instance's __dict__ dictionary, similar to the standard library implementation. ```APIDOC ## cached_property(func) ### Description This decorator transforms a method into a property whose value is computed once and cached in the instance's __dict__. It is functionally equivalent to the standard library's cached_property. ### Parameters - **func** (callable) - The method to be decorated. ### Usage ```python from propcache.api import cached_property class MyClass: @cached_property def value(self): return expensive_calculation() ``` ```