Decorators#

This module provides a collection of useful decorators for enhancing function and class behavior.

Quick Examples#

Cache Decorator#

from core_mixins.decorators import cache

@cache
def expensive_computation(n):
    return sum(range(n))

result = expensive_computation(1000)  # Computed
result = expensive_computation(1000)  # Cached

# Bounded cache — keeps the 128 most-recently used results
@cache(maxsize=128)
def fetch_record(record_id):
    return db.get(record_id)

Count Calls Decorator#

from core_mixins.decorators import count_calls

@count_calls
def my_function():
    return "Hello"

my_function()
my_function()
print(my_function.calls_number)  # Output: 2

Retry Decorator#

from core_mixins.decorators import retry
import requests

@retry(tries=3, delay=1, backoff=2)
def fetch_data(url):
    response = requests.get(url)
    response.raise_for_status()
    return response.json()

Singleton Decorator#

from core_mixins.decorators import singleton

@singleton
class DatabaseConnection:
    def __init__(self):
        self.connection = "Connected"

db1 = DatabaseConnection()
db2 = DatabaseConnection()
assert db1 is db2  # Same instance

With Time Out Decorator#

from core_mixins.decorators import with_timeout

@with_timeout(timeout=2)
def long_task_that_times_out():
    """Example function that will timeout."""
    time.sleep(5)
    return "Task completed"

@with_timeout(timeout=5)
def long_task_that_succeeds():
    """Example function that completes before timeout."""
    time.sleep(2)
    return "Task completed successfully"

if __name__ == "__main__":
    # Test 1: Function that completes successfully
    try:
        result = long_task_that_succeeds()
        print(f"Success: {result}")
    except TimeoutError as e:
        print(f"Timeout: {e}")

    # Test 2: Function that times out
    try:
        result = long_task_that_times_out()
        print(f"Success: {result}")
    except TimeoutError as e:
        print(f"Timeout: {e}")

Important:

Use with caution when combining with asyncio because the decorator spawns a subprocess and uses inter-process communication to pass data. When combined with functions like run_in_executor, you could face issues if you are working with large files due:

  • Pickling large data.

  • Fork in async context.

  • Double overhead.

Alternative

loop = asyncio.get_event_loop()
try:
    return await asyncio.wait_for(
        loop.run_in_executor(None, the_function),
        timeout=timeout
    )

except asyncio.TimeoutError:
    raise TimeoutError(...)

Time Out Decorator (Signal/Linux)#

from core_mixins.decorators import with_timeout_signal

@with_timeout_signal(timeout=2)
def long_task_that_times_out():
    return "Task completed"

Timer Decorator#

from core_mixins.decorators import timer

@timer
def slow_function():
    import time
    time.sleep(1)
    return "Done"

result, elapsed = slow_function()
print(f"Took {elapsed:.2f} seconds")

API Reference#

SyncWrapper#

Async wrapper for executing async code from synchronous context.

class core_mixins.decorators.async_.SyncWrapper(async_instance)[source]#

Bases: object

It provides an interface (mechanism) to execute async code from sync apps via a background thread that keeps a single event loop alive, avoiding asyncio.run() which creates and destroys an event loop each time and becomes expensive for many method calls.

class Test:
    def sync_method(self) -> str:
        return self.__class__.__name__

    async def testing(self) -> bool:
        await sleep(0)
        return True

sync_instance = SyncWrapper(instance)
assert Test.__name__, sync_instance.sync_method()
assert sync_instance.testing(), True
sync_instance.close()
__init__(async_instance)[source]#
close()[source]#

Close the async instance and stop the event loop.

cache#

Caching decorator for memoizing function results.

core_mixins.decorators.cache.cache(fcn: Callable | None = None, *, maxsize: int | None = None) → Callable[source]#

It maintains a cache of previous function call results that can be used to optimize the performance…

Parameters:

  • fcn – The function being decorated.

  • maxsize – Maximum number of entries to keep. When the limit is reached the least-recently-used entry is evicted. None (default) means the cache is unbounded.

Returns:

The wrapped function.

Return type:

Callable

count_calls#

Decorators for counting function call invocations.

core_mixins.decorators.count_calls.count_calls(fcn: Callable) → _CountWrapper[source]#

It provides the mechanism to count invocations…

Parameters:

fcn (Callable) – The function being decorated.

Returns:

The wrapped function with calls_number attribute.

Return type:

_CountWrapper

repeat#

Repeat decorator for executing functions multiple times.

core_mixins.decorators.repeat.repeat(fcn: Callable | None = None, *, times: int = 2) → Callable[source]#

Repeat n times the function and return the list of returned values…

Parameters:

  • fcn (Callable) – The function being decorated.

  • times (int) – Number of times the function will be invoked.

Returns:

The wrapped function.

Return type:

Callable

retry#

Retry decorator with exponential backoff for handling exceptions.

core_mixins.decorators.retry.retry(fcn: ~typing.Callable | None = None, tries: int = 3, delay: int = 1, backoff: int = 2, exceptions: ~typing.Tuple[~typing.Type[BaseException], ...] = (<class 'Exception'>,), logger: ~logging.Logger | None = None) → Callable[source]#

It retries the decorated function using an exponential backoff in case of errors (exceptions) in the execution…

Parameters:

  • fcn (Callable) – The function being decorated.

  • tries (int) – Number of retries.

  • delay (int) – Delay in seconds between invocations.

  • backoff (int) – Exponential backoff to used between retries.

  • exceptions (Callable) – Exceptions to capture for the retries. If the raised exception is not here, no retries are performed.

  • logger (Callable) – The function being decorated.

Returns:

The wrapped function.

Return type:

Callable

singleton#

Singleton pattern decorator for ensuring single instance classes.

core_mixins.decorators.singleton.singleton(cls: Type) → Callable[source]#

Make a class a Singleton class (only one instance)

Parameters:

cls (Type) – The class being decorated.

Returns:

The wrapped function.

Return type:

Callable

with_timeout#

This decorator ensures a function does not run for more than a certain time, otherwise an error is raised.

core_mixins.decorators.timeout.with_timeout(fcn: Callable | None = None, timeout: float = 10) → Callable[source]#

It executes a function that will time out after the specified value.

Important: Use with caution when combining with asyncio because the decorator spawns a subprocess and uses multiprocessing.Queue to pass data. When combined with functions like run_in_executor, you could face issues if you are working with large files due:

  • Pickling large data.

  • Fork in async context.

  • Double overhead.

Alternative

loop = asyncio.get_event_loop()
try:
    return await asyncio.wait_for(
        loop.run_in_executor(None, the_function),
        timeout=timeout
    )

except asyncio.TimeoutError:
    raise TimeoutError(...)
Parameters:

  • fcn – The function being decorated.

  • timeout – The seconds before time out.

Returns:

The wrapped function.

with_timeout_signal#

This decorator (alternative to with_timeout) ensures a function does not run for more than a certain amount of seconds before an error (TimeoutError) is raised.

core_mixins.decorators.timeout_signal.with_timeout_signal(fcn: Callable | None = None, timeout: float = 10) → Callable[source]#

It executes a function that will time out after the specified amount of seconds using signal. Could be an alternative for with_timeout because does not require spanning another process or the use of a queue.

Caveats:

  • Only works on Unix/Linux systems (SIGALRM not available on Windows).

  • Must be called from the main thread (signal.alarm limitation).

  • Not re-entrant: nested or concurrent decorated functions will interfere.

  • Not compatible with async functions or async contexts.

  • May conflict with other code using SIGALRM.

Parameters:

  • fcn – The function being decorated.

  • timeout – The seconds before time out.

Returns:

The wrapped function.

Raises:

  • TimeoutError – If the decorated function exceeds the timeout duration.

  • RuntimeError – If used on Windows or from a non-main thread.

timer#

Timer decorator for measuring function execution time.

core_mixins.decorators.timer.timer(fcn: Callable | None = None) → Callable[source]#

Using this decorator you will get a tuple in the form of: result, execution_time…

Parameters:

fcn (Callable) – The function being decorated.

Returns:

The wrapped function.

Return type:

Callable