pyfunctools.decorators package

no-index:

pyfunctools.decorators.async_decorator(func: Callable[[Any], Any])[source]

A decorator to run a synchronous function asynchronously using asyncio.

This decorator converts a blocking function into an async function by running it in an executor. It uses the asyncio event loop to handle the execution.

Parameters:: func (callable) – The synchronous function to be decorated.
Returns:: An async function that wraps the original function.
Return type:: callable

Example

>>> import time
>>>
>>> @async_decorator
... def blocking_function(x):
...     import time
...     time.sleep(2)
...     return x * 2
>>>
>>> async def main():
...     result = await blocking_function(5)
...     print(result)
>>>
>>> asyncio.run(main())
10

pyfunctools.decorators.retry_decorator(retries=3, delay=2)[source]

A decorator to retry a function execution upon failure.

This decorator retries the execution of the decorated function a specified number of times, with a delay between each attempt, in case of an exception.

Parameters:

retries (int) – The number of retry attempts. Default is 3.
delay (int) – The delay between retries in seconds. Default is 2.

Returns:

A function that retries the original function upon failure.

Return type:

callable

Example

>>> @retry_decorator(retries=5, delay=1)
... def flaky_function():
...     import random
...     if random.choice([True, False]):
...         raise ValueError("Oops, something went wrong!")
...     return "Success!"
>>>
>>> print(flaky_function())
"Success!" # (or an exception after retries)

pyfunctools.decorators.threaded_decorator(func)[source]

A decorator to run a function in a separate thread.

This decorator uses a ThreadPoolExecutor to run the decorated function in a separate thread, allowing for concurrent execution.

Parameters:: func (callable) – The function to be decorated.
Returns:: A function that runs the original function in a thread.
Return type:: callable

Example

>>> @threaded_decorator
... def slow_function(x):
...     import time
...     time.sleep(2)
...     return x * 2
>>>
>>> print(slow_function(5))
10

pyfunctools.decorators.timeout_decorator(seconds: int) → Callable[[C], C][source]

A decorator that raises a TimeoutError if the decorated function takes longer than a specified time to execute.

Parameters:: seconds (int) – The maximum allowed time for the function to execute.
Returns:: The decorated function with timeout control applied.
Return type:: Callable

Examples

>>> @timeout_decorator(1)
... def func_test(say: str):
...     pass
>>>
>>> func_test('Something')
Function func_test timed out after 1 seconds
>>> @timeout_decorator(1)
... class MyClass:
...     def __init__(self, say: str):
...         pass
>>>
>>> MyClass('Something')
Class MyClass timed out after 1 seconds

pyfunctools.decorators.timing_decorator(log_func: Callable[[str, float, float], Any] = None)[source]

A decorator to measure the execution time of a function.

This decorator measures the time taken by the decorated function to execute and logs the start time, end time, and duration. It can use a custom logging function if provided; otherwise, it defaults to printing.

Parameters:: log_func (t.Callable[[str, float, float], t.Any], optional) – A custom logging function that accepts three arguments: the function name, start time, and end time. If not provided, it defaults to printing the information.
Returns:: A function that measures the execution time of the original function.
Return type:: callable

Example

>>> def custom_logger(func_name, start_time, end_time):
...     print(f'{func_name} started at * and ended at *, taking * seconds')
>>>
>>> @timing_decorator(log_func=custom_logger)
... def example_function(x):
...     time.sleep(2)
...     return x * 2
>>>
>>> print(example_function(5))
example_function started at * and ended at *, taking * seconds
10

pyfunctools.decorators package

Submodules