idmtools.utils.decorators module

class idmtools.utils.decorators.abstractstatic(function)

Bases: staticmethod

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

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

class idmtools.utils.decorators.SingletonDecorator(klass)

Bases: object

Wraps a class in a singleton decorator.

Example

In the below example, we would print out 99 since z is referring to the same object as x:

class Thing:
    y = 14
Thing = SingletonDecorator(Thing)
x = Thing()
x.y = 99
z = Thing()
print(z.y)

class idmtools.utils.decorators.LoadOnCallSingletonDecorator(klass)

Bases: object

Additional class decorator that creates a singleton instance only when a method or attribute is accessed. This is useful for expensive tasks like loading plugin factories that should only be executed when finally needed and not on declaration.

Examples

import time
class ExpensiveFactory:
    def __init__():
        time.sleep(1000)
        self.items = ['a', 'b', 'c']
    def get_items():
        return self.items

ExpensiveFactory = LoadOnCallSingletonDecorator(ExpensiveFactory)
ExpensiveFactory.get_items()

ensure_created()

idmtools.utils.decorators.cache_for(ttl=datetime.timedelta(seconds=60))

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

Adds a CLI spinner to a function 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: Optional[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

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)

parallelize(func)

join()

get_results(futures, ordered=False)