mafw.decorators

The module provides some general decorator utilities that are used in several parts of the code, and that can be reused by the user community.

Module Attributes

F

TypeVar for generic function.

P

TypeVar for generic processor.

single_loop(cls)

A decorator shortcut to define a single execution processor.

for_loop(cls)

A decorator shortcut to define a for loop execution processor.

while_loop(cls)

A decorator shortcut to define a while loop execution processor.

Functions

class_depends_on_optional(module_name[, ...])

Class decorator factory to check if module module_name is available.

database_required(cls)

Modify the processor start method to check if a database object exists.

depends_on_optional(module_name[, raise_ex, ...])

Function decorator to check if module_name is available.

execution_workflow([loop_type])

A decorator factory for the definition of the looping strategy.

for_loop(cls)

A decorator shortcut to define a for loop execution processor.

orphan_protector(cls)

A class decorator to modify the init method of a Processor so that the remove_orphan_files is set to False and no orphan files will be removed.

processor_depends_on_optional(module_name[, ...])

Class decorator factory to check if module module_name is available.

single_loop(cls)

A decorator shortcut to define a single execution processor.

singleton(cls)

Make a class a Singleton class (only one instance)

suppress_warnings(func)

Decorator to suppress warnings during the execution of a test function.

while_loop(cls)

A decorator shortcut to define a while loop execution processor.

class mafw.decorators.F

TypeVar for generic function.

alias of TypeVar(‘F’, bound=Callable[[…], object])

class mafw.decorators.P

TypeVar for generic processor.

alias of TypeVar(‘P’, bound=Processor)

mafw.decorators.class_depends_on_optional(module_name: str, raise_ex: bool = False, warn: bool = True) Callable[[Type[Any]], Type[Any]][source]

Class decorator factory to check if module module_name is available.

It checks if all the optional modules listed in module_name separated by a ‘;’ can be found.

If all modules are found, then the class is returned as it is.

If at least one module is not found:
  • and raise_ex is True, an ImportError exception is raised and the user is responsible to deal with it.

  • if raise_ex is False, instead of returning the class, a new empty class is returned.

  • depending on the value of warn, the user will be informed with a warning message or not.

Parameters:
  • module_name (str) – The optional module(s) from which the class depends on. A “;” separated list of modules can also be provided.

  • raise_ex (bool, Optional) – Flag to raise an exception if module_name not found, defaults to False.

  • warn (bool, Optional) – Flag to display a warning message if module_name is not found, defaults to True.

Returns:

The wrapped class.

Return type:

type(object)

Raises:

ImportError – if module_name is not found and raise_ex is True.

mafw.decorators.database_required(cls)[source]

Modify the processor start method to check if a database object exists.

This decorator must be applied to processors requiring a database connection.

Parameters:

cls – A Processor class.

mafw.decorators.depends_on_optional(module_name: str, raise_ex: bool = False, warn: bool = True) Callable[[F], Callable[[...], Any]][source]

Function decorator to check if module_name is available.

If module_name is found, then returns the wrapped function. If not, its behavior depends on the raise_ex and warn_only values. If raise_ex is True, then an ImportError exception is raised. If it is False and warn is True, then a warning message is displayed but no exception is raised. If they are both False, then function is silently skipped.

If raise_ex is True, the value of warn is not taken into account.

Typical usage

The user should decorate functions or class methods when they cannot be executed without the optional library. In the specific case of Processor subclass, where the class itself can be created also without the missing library, but it is required somewhere in the processor execution, then the user is suggested to decorate the execute method with this decorator.

Parameters:
  • module_name (str) – The optional module(s) from which the function depends on. A “;” separated list of modules can also be provided.

  • raise_ex (bool, Optional) – Flag to raise an exception if module_name is not found, defaults to False.

  • warn (bool, Optional) – Flag to display a warning message if module_name is not found, default to True.

Returns:

The wrapped function

Return type:

Callable

Raises:

ImportError – if module_name is not found and raise_ex is True.

mafw.decorators.execution_workflow(loop_type: LoopType | str = LoopType.ForLoop)[source]

A decorator factory for the definition of the looping strategy.

This decorator factory must be applied to Processor subclasses to modify their value of loop_type in order to change the execution workflow.

See single_loop(), for_loop() and while_loop() decorator shortcuts.

Parameters:

loop_type (LoopType | str, Optional) – The type of execution workflow requested for the decorated class. Defaults to LoopType.ForLoop.

mafw.decorators.for_loop(cls)

A decorator shortcut to define a for loop execution processor.

mafw.decorators.orphan_protector(cls)[source]

A class decorator to modify the init method of a Processor so that the remove_orphan_files is set to False and no orphan files will be removed.

mafw.decorators.processor_depends_on_optional(module_name: str, raise_ex: bool = False, warn: bool = True) Callable[[Type[P]], Type[Processor]][source]

Class decorator factory to check if module module_name is available.

It checks if all the optional modules listed in module_name separated by a ‘;’ can be found.

If all modules are found, then the class is returned as it is.

If at least one module is not found:
  • and raise_ex is True, an ImportError exception is raised and the user is responsible to deal with it.

  • if raise_ex is False, instead of returning the class, the Processor is returned.

  • depending on the value of warn, the user will be informed with a warning message or not.

Typical usage

The user should decorate Processor subclasses everytime the optional module is required in their __init__ method. Should the check on the optional module have a positive outcome, then the Processor subclass is returned. Otherwise, if raise_ex is False, an instance of the base Processor is returned. In this way, the returned class can still be executed without breaking the execution scheme but of course, without producing any output.

Should be possible to run the __init__ method of the class without the missing library, then the user can also follow the approach described in this other example.

Parameters:
  • module_name (str) – The optional module(s) from which the class depends on. A “;” separated list of modules can also be provided.

  • raise_ex (bool, Optional) – Flag to raise an exception if module_name not found, defaults to False.

  • warn (bool, Optional) – Flag to display a warning message if module_name is not found, defaults to True.

Returns:

The wrapped processor.

Return type:

type(Processor)

Raises:

ImportError – if module_name is not found and raise_ex is True.

mafw.decorators.single_loop(cls)

A decorator shortcut to define a single execution processor.

mafw.decorators.singleton(cls)[source]

Make a class a Singleton class (only one instance)

mafw.decorators.suppress_warnings(func: F) F[source]

Decorator to suppress warnings during the execution of a test function.

This decorator uses the warnings.catch_warnings() context manager to temporarily change the warning filter to ignore all warnings. It is useful when you want to run a test without having warnings clutter the output.

Usage:

@suppress_warnings
def test_function():
    # Your test code that might emit warnings
Parameters:

func (Callable) – The test function to be decorated.

Returns:

The wrapped function with suppressed warnings.

Return type:

Callable

mafw.decorators.while_loop(cls)

A decorator shortcut to define a while loop execution processor.