idmtools.utils.decorators module¶

Decorators defined for idmtools.

class idmtools.utils.decorators.abstractstatic(function)¶

Bases: staticmethod

A decorator for defining a method both as static and abstract.

__init__(function)¶

Initialize abstractstatic.

Parameters: function – Function to wrap as abstract

idmtools.utils.decorators.optional_decorator(decorator: Callable, condition: Union[bool, Callable[[], bool]])¶

A decorator that adds a decorator only if condition is true.

Parameters

decorator – Decorator to add
condition – Condition to determine. Condition can be a callable as well

Returns

Optionally wrapped func.

class idmtools.utils.decorators.SingletonMixin¶

Bases: object

SingletonMixin defines a singleton that can be added to any class.

As a singleton, on one instance will be made per process.

classmethod instance()¶

Return the instance of the object.

If the instance has not been created, it will be initialized before returning.

Returns: The singleton instance

idmtools.utils.decorators.cache_for(ttl=None) → Callable¶

Cache a value for a certain time period.

Parameters: ttl – Expiration of cache
Returns: Wrapper Function

idmtools.utils.decorators.optional_yaspin_load(*yargs, **ykwargs) → Callable¶

Adds a CLI spinner to a function based on conditions.

The spinner will be present if

yaspin package is present.
NO_SPINNER environment variable is not defined.

Parameters

*yargs – Arguments to pass to yaspin constructor.
**ykwargs – Keyword arguments to pass to yaspin constructor.

Examples

@optional_yaspin_load(text="Loading test", color="yellow")
def test():
    time.sleep(100)

Returns: A callable wrapper function.

class idmtools.utils.decorators.ParallelizeDecorator(queue=None, pool_type: ~typing.Optional[~typing.Type[~concurrent.futures._base.Executor]] = <class 'concurrent.futures.thread.ThreadPoolExecutor'>)¶

Bases: object

ParallelizeDecorator allows you to easily parallelize a group of code.

A simple of example would be following

Examples

op_queue = ParallelizeDecorator()

class Ops:
    op_queue.parallelize
    def heavy_op():
        time.sleep(10)

    def do_lots_of_heavy():
        futures = [self.heavy_op() for i in range(100)]
        results = op_queue.get_results(futures)

__init__(queue=None, pool_type: ~typing.Optional[~typing.Type[~concurrent.futures._base.Executor]] = <class 'concurrent.futures.thread.ThreadPoolExecutor'>)¶

Initialize our ParallelizeDecorator.

Parameters

queue – Queue to use. If not provided, one will be created.
pool_type – Pool type to use. Defaults to ThreadPoolExecutor.

parallelize(func)¶

Wrap a function in parallelization.

Parameters: func – Function to wrap with parallelization
Returns: Function wrapped with parallelization object

join()¶

Join our queue.

Returns: Join operation from queue

get_results(futures, ordered=False)¶

Get Results from our decorator.

Parameters

futures – Futures to get results from
ordered – Do we want results in order provided or as they complete. Default is as they complete which is False.

Returns

Results from all the futures.