Coverage for src / mafw / decorators.py: 99%
127 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-30 16:10 +0000
1# Copyright 2025–2026 European Union
2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
3# SPDX-License-Identifier: EUPL-1.2
4"""
5The module provides some general decorator utilities that are used in several parts of the code, and that can be
6reused by the user community.
7"""
9import functools
10import typing
11import warnings
12from importlib.util import find_spec
13from typing import Any, Callable, Type
15from mafw.enumerators import LoopType
16from mafw.mafw_errors import MissingDatabase, MissingOptionalDependency
17from mafw.processor import Processor
19F = typing.TypeVar('F', bound=typing.Callable[..., object])
20"""TypeVar for generic function."""
22# Define a TypeVar to capture Processor subclasses
23P = typing.TypeVar('P', bound=Processor)
24"""TypeVar for generic processor."""
27def suppress_warnings(func: F) -> F:
28 """
29 Decorator to suppress warnings during the execution of a test function.
31 This decorator uses the `warnings.catch_warnings()` context manager to
32 temporarily change the warning filter to ignore all warnings. It is useful
33 when you want to run a test without having warnings clutter the output.
35 Usage::
37 @suppress_warnings
38 def test_function():
39 # Your test code that might emit warnings
42 :param func: The test function to be decorated.
43 :type func: Callable
44 :return: The wrapped function with suppressed warnings.
45 :rtype: Callable
46 """
48 @functools.wraps(func)
49 def wrapper(*args: Any, **kwargs: Any) -> F:
50 with warnings.catch_warnings():
51 warnings.simplefilter('ignore')
52 return func(*args, **kwargs) # type: ignore[return-value]
54 return wrapper # type: ignore[return-value]
57@typing.no_type_check # no idea how to fix it
58def singleton(cls):
59 """Make a class a Singleton class (only one instance)"""
61 @functools.wraps(cls)
62 def wrapper_singleton(*args: Any, **kwargs: Any):
63 if wrapper_singleton.instance is None:
64 wrapper_singleton.instance = cls(*args, **kwargs)
65 return wrapper_singleton.instance
67 wrapper_singleton.instance = None
68 return wrapper_singleton
71@typing.no_type_check
72def database_required(cls):
73 """Modify the processor start method to check if a database object exists.
75 This decorator must be applied to processors requiring a database connection.
77 It operates on a processor class modifying its start method. Since the start method is also one of those for
78 which the presence of the super call is verified, this decorator is also taking care of re-applying the super
79 call wrapper.
81 :param cls: A Processor class.
82 """
83 orig_start = cls.start
85 @functools.wraps(cls.start)
86 def _start(self) -> None:
87 if self._database is None:
88 raise MissingDatabase(f'{self.name} requires an active database.')
89 orig_start(self)
91 cls.start = _start
92 if hasattr(cls, '_apply_super_call_wrappers'): 92 ↛ 95line 92 didn't jump to line 95 because the condition on line 92 was always true
93 cls._apply_super_call_wrappers()
95 return cls
98@typing.no_type_check
99def orphan_protector(cls):
100 """
101 A class decorator to modify the init method of a Processor so that the remove_orphan_files is set to False and
102 no orphan files will be removed.
103 """
104 old_init = cls.__init__
106 @functools.wraps(cls.__init__)
107 def new_init(self, *args, **kwargs):
108 old_init(self, *args, remove_orphan_files=False, **kwargs)
110 cls.__init__ = new_init
111 return cls
114@typing.no_type_check
115def execution_workflow(loop_type: LoopType | str = LoopType.ForLoop):
116 """
117 A decorator factory for the definition of the looping strategy.
119 This decorator factory must be applied to Processor subclasses to modify their value of loop_type in order to
120 change the execution workflow.
122 See :func:`single_loop`, :func:`for_loop` and :func:`while_loop` decorator shortcuts.
124 :param loop_type: The type of execution workflow requested for the decorated class. Defaults to LoopType.ForLoop.
125 :type loop_type: LoopType | str, Optional
126 """
128 def dec(cls):
129 """The class decorator."""
130 old_init = cls.__init__
132 @functools.wraps(cls.__init__)
133 def new_init(self, *args, **kwargs):
134 """The modified Processor init"""
135 old_init(self, *args, looper=loop_type, **kwargs)
137 cls.__init__ = new_init
139 return cls
141 return dec
144single_loop = execution_workflow(LoopType.SingleLoop)
145"""A decorator shortcut to define a single execution processor."""
147for_loop = execution_workflow(LoopType.ForLoop)
148"""A decorator shortcut to define a for loop execution processor."""
150while_loop = execution_workflow(LoopType.WhileLoop)
151"""A decorator shortcut to define a while loop execution processor."""
153parallel_for_loop = execution_workflow(LoopType.ParallelForLoop)
154"""A decorator shortcut to define a parallel for loop execution processor."""
156parallel_for_loop_with_queue = execution_workflow(LoopType.ParallelForLoopWithQueue)
157"""A decorator shortcut to define a parallel for loop with queue execution processor."""
160def depends_on_optional(
161 module_name: str, raise_ex: bool = False, warn: bool = True
162) -> Callable[[F], Callable[..., Any]]:
163 """
164 Function decorator to check if module_name is available.
166 If module_name is found, then returns the wrapped function. If not, its behavior depends on the raise_ex and
167 warn_only values. If raise_ex is True, then an ImportError exception is raised. If it is False and warn is
168 True, then a warning message is displayed but no exception is raised. If they are both False, then function is
169 silently skipped.
171 If raise_ex is True, the value of `warn` is not taken into account.
173 **Typical usage**
175 The user should decorate functions or class methods when they cannot be executed without the optional library.
176 In the specific case of Processor subclass, where the class itself can be created also without the missing
177 library, but it is required somewhere in the processor execution, then the user is suggested to decorate the
178 execute method with this decorator.
180 :param module_name: The optional module(s) from which the function depends on. A ";" separated list of modules can
181 also be provided.
182 :type module_name: str
183 :param raise_ex: Flag to raise an exception if module_name is not found, defaults to False.
184 :type raise_ex: bool, Optional
185 :param warn: Flag to display a warning message if module_name is not found, default to True.
186 :type warn: bool, Optional
187 :return: The wrapped function
188 :rtype: Callable
189 :raise ImportError: if module_name is not found and raise_ex is True.
190 """
192 def decorator(func: F) -> Callable[..., Any]:
193 @functools.wraps(func)
194 def wrapper(*args: Any, **kwargs: Any) -> Any:
195 all_mods_found = all(find_spec(mod.strip()) is not None for mod in module_name.split(';'))
196 if not all_mods_found:
197 msg = f'Optional dependency {module_name} not found ({func.__qualname__})'
198 if raise_ex:
199 raise ImportError(msg)
200 else:
201 if warn:
202 warnings.warn(MissingOptionalDependency(msg), stacklevel=2)
203 return None # Explicitly return None when skipping the function
204 else:
205 return func(*args, **kwargs)
207 return wrapper
209 return decorator
212def processor_depends_on_optional(
213 module_name: str, raise_ex: bool = False, warn: bool = True
214) -> Callable[[Type[P]], Type[Processor]]:
215 """
216 Class decorator factory to check if module module_name is available.
218 It checks if all the optional modules listed in `module_name` separated by a ';' can be found.
220 If all modules are found, then the class is returned as it is.
222 If at least one module is not found:
223 - and raise_ex is True, an ImportError exception is raised and the user is responsible to deal with it.
224 - if raise_ex is False, instead of returning the class, the :class:`~.Processor` is returned.
225 - depending on the value of warn, the user will be informed with a warning message or not.
227 **Typical usage**
229 The user should decorate Processor subclasses everytime the optional module is required in their __init__ method.
230 Should the check on the optional module have a positive outcome, then the Processor subclass is returned.
231 Otherwise, if raise_ex is False, an instance of the base :py:class:`~.Processor` is returned. In
232 this way, the returned class can still be executed without breaking the execution scheme but of course, without
233 producing any output.
235 Should be possible to run the __init__ method of the class without the missing library, then the user can also
236 follow the approach described in this other :func:`example <depends_on_optional>`.
238 :param module_name: The optional module(s) from which the class depends on. A ";" separated list of modules can
239 also be provided.
240 :type module_name: str
241 :param raise_ex: Flag to raise an exception if module_name not found, defaults to False.
242 :type raise_ex: bool, Optional
243 :param warn: Flag to display a warning message if module_name is not found, defaults to True.
244 :type warn: bool, Optional
245 :return: The wrapped processor.
246 :rtype: type(Processor)
247 :raise ImportError: if module_name is not found and raise_ex is True.
248 """
250 def decorator(cls: Type[P]) -> Type[Processor]:
251 """
252 The class decorator.
254 It checks if all the modules provided by the decorator factory are available on the systems.
255 If yes, then it simply returns `cls`. If no, it returns a subclass of the :class:`~.Processor`
256 after all the introspection properties have been taken from `cls`.
258 :param cls: The class being decorated.
259 :type cls: type(Processor)
260 :return: The decorated class, either cls or a subclass of :class:`~autorad.processor.Processor`.
261 :rtype: type(Processor)
262 """
264 def class_wrapper(klass: Type[Processor]) -> Type[Processor]:
265 """
266 Copy introspection properties from cls to klass.
268 :param klass: The class to be modified.
269 :type klass: class.
270 :return: The modified class.
271 :rtype: class.
272 """
273 klass.__module__ = cls.__module__
274 klass.__name__ = f'{cls.__name__} (Missing {module_name})'
275 klass.__qualname__ = cls.__qualname__
276 klass.__annotations__ = cls.__annotations__
277 klass.__doc__ = cls.__doc__
278 return klass
280 all_mods_found = all([find_spec(mod.strip()) is not None for mod in module_name.split(';')])
281 if not all_mods_found:
282 msg = f'Optional dependency {module_name} not found ({cls.__qualname__})'
283 if raise_ex:
284 raise ImportError(msg)
285 else:
286 if warn:
287 warnings.warn(MissingOptionalDependency(msg), stacklevel=2)
289 # We subclass the basic processor.
290 class NewClass(Processor):
291 pass
293 # The class wrapper is copying introspection properties from the cls to the NewClass
294 new_class = class_wrapper(NewClass)
296 else:
297 new_class = cls
298 return new_class
300 return decorator
303def class_depends_on_optional(
304 module_name: str, raise_ex: bool = False, warn: bool = True
305) -> Callable[[Type[Any]], Type[Any]]:
306 """
307 Class decorator factory to check if module module_name is available.
309 It checks if all the optional modules listed in `module_name` separated by a ';' can be found.
311 If all modules are found, then the class is returned as it is.
313 If at least one module is not found:
314 - and raise_ex is True, an ImportError exception is raised and the user is responsible to deal with it.
315 - if raise_ex is False, instead of returning the class, a new empty class is returned.
316 - depending on the value of warn, the user will be informed with a warning message or not.
318 :param module_name: The optional module(s) from which the class depends on. A ";" separated list of modules can
319 also be provided.
320 :type module_name: str
321 :param raise_ex: Flag to raise an exception if module_name not found, defaults to False.
322 :type raise_ex: bool, Optional
323 :param warn: Flag to display a warning message if module_name is not found, defaults to True.
324 :type warn: bool, Optional
325 :return: The wrapped class.
326 :rtype: type(object)
327 :raise ImportError: if module_name is not found and raise_ex is True.
328 """
330 def decorator(cls: Type[Any]) -> Type[Any]:
331 """
332 The class decorator.
334 It checks if all the modules provided by the decorator factory are available on the systems.
335 If yes, then it simply returns `cls`. If no, it returns a subclass of the cls bases.
336 after all the introspection properties have been taken from `cls`.
338 :param cls: The class being decorated.
339 :type cls: type(cls)
340 :return: The decorated class, either cls or a subclass of cls.
341 :rtype: type(cls)
342 """
344 def class_wrapper(klass: Type[Any]) -> Type[Any]:
345 """
346 Copy introspection properties from cls to klass.
348 :param klass: The class to be modified.
349 :type klass: class.
350 :return: The modified class.
351 :rtype: class.
352 """
353 klass.__module__ = cls.__module__
354 klass.__name__ = f'{cls.__name__} (Missing {module_name})'
355 klass.__qualname__ = cls.__qualname__
356 klass.__annotations__ = cls.__annotations__
357 klass.__doc__ = cls.__doc__
358 return klass
360 all_mods_found = all([find_spec(mod.strip()) is not None for mod in module_name.split(';')])
361 if not all_mods_found:
362 msg = f'Optional dependency {module_name} not found ({cls.__qualname__})'
363 if raise_ex:
364 raise ImportError(msg)
365 else:
366 if warn:
367 warnings.warn(MissingOptionalDependency(msg), stacklevel=2)
369 # we subclass the original class.
370 class NewClass(*cls.__bases__): # type: ignore
371 pass
373 # the class wrapper is copying introspection properties from the cls to the NewClass
374 new_class = class_wrapper(NewClass)
376 else:
377 new_class = cls
378 return new_class
380 return decorator