crantpy.utils.decorators module

This module provides decorators for caching and other utilities.

crantpy.utils.decorators.cached_result(cache_name, max_age=7200, key_fn=None, should_cache_fn=None, validate_cache_fn=None)

Decorator for caching function results.

WARNING: This decorator is not thread-safe. It is recommended to use threading.Lock() to ensure thread safety when using this decorator in a multi-threaded environment.

This decorator provides a flexible caching mechanism for function results. It supports custom cache keys, validation, and conditional caching, making it suitable for a variety of use cases.

The cache stores entries in a dictionary structure: {

‘result’: original_function_result, ‘metadata’: {

‘_created_at’: timestamp

}

}

This approach avoids modifying the original result objects directly, ensuring compatibility with immutable types.

Parameters:

• cache_name (str) – Name of the global cache to use. This is used to group cached results under a specific namespace.

• max_age (int, default MAXIMUM_CACHE_DURATION) – Maximum age of cached result in seconds. Cached results older than this duration are considered stale and will be refreshed.

• key_fn (callable, optional) – Function to generate a unique cache key based on the function’s arguments. Defaults to using the first positional argument or the ‘dataset’ keyword argument. If the function returns None, an error will be raised.

• should_cache_fn (callable, optional) – Function to determine whether the result of the function should be cached. It takes the function result and arguments as input and returns a boolean.

• validate_cache_fn (callable, optional) – Function to validate if a cached result is still valid beyond the age check. It takes the cached result and the function arguments as input and returns a boolean.

Returns:

The decorated function with caching capabilities.

Return type:

callable

Examples

>>> # Basic Caching:
>>> @cached_result(cache_name="example_cache")
>>> def expensive_function(x):
>>>     return x ** 2

>>> # Custom Cache Key:
>>> @cached_result(cache_name="example_cache", key_fn=lambda x: f"key_{x}")
>>> def expensive_function(x):
>>>     return x ** 2

>>> # Conditional Caching:
>>> @cached_result(cache_name="example_cache", should_cache_fn=lambda result, *args: result > 10)
>>> def expensive_function(x):
>>>     return x ** 2

>>> # Cache Validation:
>>> def validate_cache(result, *args):
>>>     return result is not None

>>> @cached_result(cache_name="example_cache", validate_cache_fn=validate_cache)
>>> def expensive_function(x):
>>>     return x ** 2

Notes

• The decorated function gains a clear_cache method to manually clear the cache for the specified cache_name.

• The check_stale parameter can be used to skip staleness checks when calling the decorated function.

crantpy.utils.decorators.clear_global_cache(cache_name)

Clear a named global cache.

Parameters:

cache_name (str) – Name of the cache to clear

Return type:

None

crantpy.utils.decorators.get_global_cache(cache_name)

Get a named global cache dictionary.

Parameters:

cache_name (str) – Name of the cache to retrieve

Returns:

The requested cache dictionary

Return type:

dict

crantpy.utils.decorators.inject_dataset(allowed=None, disallowed=None, param_name='dataset')

Inject current default dataset.

Parameters:

• allowed (List[str] or str, optional) – List of allowed datasets or a single allowed dataset.

• disallowed (List[str] or str, optional) – List of disallowed datasets or a single disallowed dataset.

• param_name (str, default 'dataset') – Name of the parameter to inject the dataset into.

Returns:

Decorator function that injects the dataset.

Return type:

Callable

crantpy.utils.decorators.parse_neuroncriteria(allow_empty=False)

Parse all NeuronCriteria arguments into an array of root IDs.

This decorator automatically converts any NeuronCriteria objects in function arguments to arrays of root IDs. It uses type checking by class name to avoid circular imports.

Parameters:

allow_empty (bool, default False) – Whether to allow the NeuronCriteria to not match any neurons.

Returns:

Decorator function that processes NeuronCriteria arguments.

Return type:

Callable

Examples

>>> @parse_neuroncriteria()
>>> def process_neurons(neurons):
>>>     # neurons will be an array of root IDs
>>>     return neurons
>>>
>>> # Can be called with a NeuronCriteria object
>>> result = process_neurons(NeuronCriteria(cell_class='example'))

crantpy.utils.decorators.retry_func(retries=5, cooldown=2)

Retry decorator for functions on HTTPError. This also suppresses UserWarnings (commonly raised by l2 cache requests) :param cooldown: Cooldown period in seconds between attempts. :type cooldown: int | float :param retries: Number of retries before we give up. Every subsequent retry

will delay by an additional retry.