Coverage for src/mafw/decorators.py: 99%
128 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-26 09:13 +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 collections.abc import Callable
13from importlib.util import find_spec
14from typing import Any
16from mafw.enumerators import LoopType
17from mafw.mafw_errors import MissingDatabase, MissingOptionalDependency
18from mafw.processor import Processor
20F = typing.TypeVar('F', bound=typing.Callable[..., object])
21"""TypeVar for generic function."""
23# Define a TypeVar to capture Processor subclasses
24P = typing.TypeVar('P', bound=Processor)
25"""TypeVar for generic processor."""
28def suppress_warnings(func: F) -> F:
29 """
30 Decorator to suppress warnings during the execution of a test function.
32 This decorator uses the `warnings.catch_warnings()` context manager to
33 temporarily change the warning filter to ignore all warnings. It is useful
34 when you want to run a test without having warnings clutter the output.
36 Usage::
38 @suppress_warnings
39 def test_function():
40 # Your test code that might emit warnings
43 :param func: The test function to be decorated.
44 :type func: Callable
45 :return: The wrapped function with suppressed warnings.
46 :rtype: Callable
47 """
49 @functools.wraps(func)
50 def wrapper(*args: Any, **kwargs: Any) -> F:
51 with warnings.catch_warnings():
52 warnings.simplefilter('ignore')
53 return func(*args, **kwargs) # type: ignore[return-value]
55 return wrapper # type: ignore[return-value]
58@typing.no_type_check # no idea how to fix it
59def singleton(cls):
60 """Make a class a Singleton class (only one instance)"""
62 @functools.wraps(cls)
63 def wrapper_singleton(*args: Any, **kwargs: Any):
64 if wrapper_singleton.instance is None:
65 wrapper_singleton.instance = cls(*args, **kwargs)
66 return wrapper_singleton.instance
68 wrapper_singleton.instance = None
69 return wrapper_singleton
72@typing.no_type_check
73def database_required(cls):
74 """Modify the processor start method to check if a database object exists.
76 This decorator must be applied to processors requiring a database connection.
78 It operates on a processor class modifying its start method. Since the start method is also one of those for
79 which the presence of the super call is verified, this decorator is also taking care of re-applying the super
80 call wrapper.
82 :param cls: A Processor class.
83 """
84 orig_start = cls.start
86 @functools.wraps(cls.start)
87 def _start(self) -> None:
88 if self._database is None:
89 raise MissingDatabase(f'{self.name} requires an active database.')
90 orig_start(self)
92 cls.start = _start
93 if hasattr(cls, '_apply_super_call_wrappers'): 93 ↛ 96line 93 didn't jump to line 96 because the condition on line 93 was always true
94 cls._apply_super_call_wrappers()
96 return cls
99@typing.no_type_check
100def orphan_protector(cls):
101 """
102 A class decorator to modify the init method of a Processor so that the remove_orphan_files is set to False and
103 no orphan files will be removed.
104 """
105 old_init = cls.__init__
107 @functools.wraps(cls.__init__)
108 def new_init(self, *args, **kwargs):
109 old_init(self, *args, remove_orphan_files=False, **kwargs)
111 cls.__init__ = new_init
112 return cls
115@typing.no_type_check
116def execution_workflow(loop_type: LoopType | str = LoopType.ForLoop):
117 """
118 A decorator factory for the definition of the looping strategy.
120 This decorator factory must be applied to Processor subclasses to modify their value of loop_type in order to
121 change the execution workflow.
123 See :func:`single_loop`, :func:`for_loop` and :func:`while_loop` decorator shortcuts.
125 :param loop_type: The type of execution workflow requested for the decorated class. Defaults to LoopType.ForLoop.
126 :type loop_type: LoopType | str, Optional
127 """
129 def dec(cls):
130 """The class decorator."""
131 old_init = cls.__init__
133 @functools.wraps(cls.__init__)
134 def new_init(self, *args, **kwargs):
135 """The modified Processor init"""
136 old_init(self, *args, looper=loop_type, **kwargs)
138 cls.__init__ = new_init
140 return cls
142 return dec
145single_loop = execution_workflow(LoopType.SingleLoop)
146"""A decorator shortcut to define a single execution processor."""
148for_loop = execution_workflow(LoopType.ForLoop)
149"""A decorator shortcut to define a for loop execution processor."""
151while_loop = execution_workflow(LoopType.WhileLoop)
152"""A decorator shortcut to define a while loop execution processor."""
154parallel_for_loop = execution_workflow(LoopType.ParallelForLoop)
155"""A decorator shortcut to define a parallel for loop execution processor."""
157parallel_for_loop_with_queue = execution_workflow(LoopType.ParallelForLoopWithQueue)
158"""A decorator shortcut to define a parallel for loop with queue execution processor."""
161def depends_on_optional(
162 module_name: str, raise_ex: bool = False, warn: bool = True
163) -> Callable[[F], Callable[..., Any]]:
164 """
165 Function decorator to check if module_name is available.
167 If module_name is found, then returns the wrapped function. If not, its behaviour depends on the raise_ex and
168 warn_only values. If raise_ex is True, then an ImportError exception is raised. If it is False and warn is
169 True, then a warning message is displayed but no exception is raised. If they are both False, then function is
170 silently skipped.
172 If raise_ex is True, the value of `warn` is not taken into account.
174 **Typical usage**
176 The user should decorate functions or class methods when they cannot be executed without the optional library.
177 In the specific case of Processor subclass, where the class itself can be created also without the missing
178 library, but it is required somewhere in the processor execution, then the user is suggested to decorate the
179 execute method with this decorator.
181 :param module_name: The optional module(s) from which the function depends on. A ";" separated list of modules can
182 also be provided.
183 :type module_name: str
184 :param raise_ex: Flag to raise an exception if module_name is not found, defaults to False.
185 :type raise_ex: bool, Optional
186 :param warn: Flag to display a warning message if module_name is not found, default to True.
187 :type warn: bool, Optional
188 :return: The wrapped function
189 :rtype: Callable
190 :raise ImportError: if module_name is not found and raise_ex is True.
191 """
193 def decorator(func: F) -> Callable[..., Any]:
194 @functools.wraps(func)
195 def wrapper(*args: Any, **kwargs: Any) -> Any:
196 all_mods_found = all(find_spec(mod.strip()) is not None for mod in module_name.split(';'))
197 if not all_mods_found:
198 msg = f'Optional dependency {module_name} not found ({func.__qualname__})'
199 if raise_ex:
200 raise ImportError(msg)
201 else:
202 if warn:
203 warnings.warn(MissingOptionalDependency(msg), stacklevel=2)
204 return None # Explicitly return None when skipping the function
205 else:
206 return func(*args, **kwargs)
208 return wrapper
210 return decorator
213def processor_depends_on_optional(
214 module_name: str, raise_ex: bool = False, warn: bool = True
215) -> Callable[[type[P]], type[Processor]]:
216 """
217 Class decorator factory to check if module module_name is available.
219 It checks if all the optional modules listed in `module_name` separated by a ';' can be found.
221 If all modules are found, then the class is returned as it is.
223 If at least one module is not found:
224 - and raise_ex is True, an ImportError exception is raised and the user is responsible to deal with it.
225 - if raise_ex is False, instead of returning the class, the :class:`~.processor.Processor` is returned.
226 - depending on the value of warn, the user will be informed with a warning message or not.
228 **Typical usage**
230 The user should decorate Processor subclasses everytime the optional module is required in their __init__ method.
231 Should the check on the optional module have a positive outcome, then the Processor subclass is returned.
232 Otherwise, if raise_ex is False, an instance of the base :py:class:`~.processor.Processor` is returned. In
233 this way, the returned class can still be executed without breaking the execution scheme but of course, without
234 producing any output.
236 Should be possible to run the __init__ method of the class without the missing library, then the user can also
237 follow the approach described in this other :func:`example <depends_on_optional>`.
239 :param module_name: The optional module(s) from which the class depends on. A ";" separated list of modules can
240 also be provided.
241 :type module_name: str
242 :param raise_ex: Flag to raise an exception if module_name not found, defaults to False.
243 :type raise_ex: bool, Optional
244 :param warn: Flag to display a warning message if module_name is not found, defaults to True.
245 :type warn: bool, Optional
246 :return: The wrapped processor.
247 :rtype: type(processor.Processor)
248 :raise ImportError: if module_name is not found and raise_ex is True.
249 """
251 def decorator(cls: type[P]) -> type[Processor]:
252 """
253 The class decorator.
255 It checks if all the modules provided by the decorator factory are available on the systems.
256 If yes, then it simply returns `cls`. If no, it returns a subclass of the :class:`~.processor.Processor`
257 after all the introspection properties have been taken from `cls`.
259 :param cls: The class being decorated.
260 :type cls: type(Processor)
261 :return: The decorated class, either cls or a subclass of :class:`~.processor.Processor`.
262 :rtype: type(Processor)
263 """
265 def class_wrapper(klass: type[Processor]) -> type[Processor]:
266 """
267 Copy introspection properties from cls to klass.
269 :param klass: The class to be modified.
270 :type klass: class.
271 :return: The modified class.
272 :rtype: class.
273 """
274 klass.__module__ = cls.__module__
275 klass.__name__ = f'{cls.__name__} (Missing {module_name})'
276 klass.__qualname__ = cls.__qualname__
277 klass.__annotations__ = cls.__annotations__
278 klass.__doc__ = cls.__doc__
279 return klass
281 all_mods_found = all([find_spec(mod.strip()) is not None for mod in module_name.split(';')])
282 if not all_mods_found:
283 msg = f'Optional dependency {module_name} not found ({cls.__qualname__})'
284 if raise_ex:
285 raise ImportError(msg)
286 else:
287 if warn:
288 warnings.warn(MissingOptionalDependency(msg), stacklevel=2)
290 # We subclass the basic processor.
291 class NewClass(Processor):
292 pass
294 # The class wrapper is copying introspection properties from the cls to the NewClass
295 new_class = class_wrapper(NewClass)
297 else:
298 new_class = cls
299 return new_class
301 return decorator
304def class_depends_on_optional(
305 module_name: str, raise_ex: bool = False, warn: bool = True
306) -> Callable[[type[Any]], type[Any]]:
307 """
308 Class decorator factory to check if module module_name is available.
310 It checks if all the optional modules listed in `module_name` separated by a ';' can be found.
312 If all modules are found, then the class is returned as it is.
314 If at least one module is not found:
315 - and raise_ex is True, an ImportError exception is raised and the user is responsible to deal with it.
316 - if raise_ex is False, instead of returning the class, a new empty class is returned.
317 - depending on the value of warn, the user will be informed with a warning message or not.
319 :param module_name: The optional module(s) from which the class depends on. A ";" separated list of modules can
320 also be provided.
321 :type module_name: str
322 :param raise_ex: Flag to raise an exception if module_name not found, defaults to False.
323 :type raise_ex: bool, Optional
324 :param warn: Flag to display a warning message if module_name is not found, defaults to True.
325 :type warn: bool, Optional
326 :return: The wrapped class.
327 :rtype: type(object)
328 :raise ImportError: if module_name is not found and raise_ex is True.
329 """
331 def decorator(cls: type[Any]) -> type[Any]:
332 """
333 The class decorator.
335 It checks if all the modules provided by the decorator factory are available on the systems.
336 If yes, then it simply returns `cls`. If no, it returns a subclass of the cls bases.
337 after all the introspection properties have been taken from `cls`.
339 :param cls: The class being decorated.
340 :type cls: type(cls)
341 :return: The decorated class, either cls or a subclass of cls.
342 :rtype: type(cls)
343 """
345 def class_wrapper(klass: type[Any]) -> type[Any]:
346 """
347 Copy introspection properties from cls to klass.
349 :param klass: The class to be modified.
350 :type klass: class.
351 :return: The modified class.
352 :rtype: class.
353 """
354 klass.__module__ = cls.__module__
355 klass.__name__ = f'{cls.__name__} (Missing {module_name})'
356 klass.__qualname__ = cls.__qualname__
357 klass.__annotations__ = cls.__annotations__
358 klass.__doc__ = cls.__doc__
359 return klass
361 all_mods_found = all([find_spec(mod.strip()) is not None for mod in module_name.split(';')])
362 if not all_mods_found:
363 msg = f'Optional dependency {module_name} not found ({cls.__qualname__})'
364 if raise_ex:
365 raise ImportError(msg)
366 else:
367 if warn:
368 warnings.warn(MissingOptionalDependency(msg), stacklevel=2)
370 # we subclass the original class.
371 class NewClass(*cls.__bases__): # type: ignore
372 pass
374 # the class wrapper is copying introspection properties from the cls to the NewClass
375 new_class = class_wrapper(NewClass)
377 else:
378 new_class = cls
379 return new_class
381 return decorator