Coverage for src / mafw / decorators.py: 100%
121 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-09 09:08 +0000
1# Copyright 2025 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 :param cls: A Processor class.
78 """
79 orig_start = cls.start
81 @functools.wraps(cls.start)
82 def _start(self) -> None:
83 if self._database is None:
84 raise MissingDatabase(f'{self.name} requires an active database.')
85 orig_start(self)
87 cls.start = _start
89 return cls
92@typing.no_type_check
93def orphan_protector(cls):
94 """
95 A class decorator to modify the init method of a Processor so that the remove_orphan_files is set to False and
96 no orphan files will be removed.
97 """
98 old_init = cls.__init__
100 @functools.wraps(cls.__init__)
101 def new_init(self, *args, **kwargs):
102 old_init(self, *args, remove_orphan_files=False, **kwargs)
104 cls.__init__ = new_init
105 return cls
108@typing.no_type_check
109def execution_workflow(loop_type: LoopType | str = LoopType.ForLoop):
110 """
111 A decorator factory for the definition of the looping strategy.
113 This decorator factory must be applied to Processor subclasses to modify their value of loop_type in order to
114 change the execution workflow.
116 See :func:`single_loop`, :func:`for_loop` and :func:`while_loop` decorator shortcuts.
118 :param loop_type: The type of execution workflow requested for the decorated class. Defaults to LoopType.ForLoop.
119 :type loop_type: LoopType | str, Optional
120 """
122 def dec(cls):
123 """The class decorator."""
124 old_init = cls.__init__
126 @functools.wraps(cls.__init__)
127 def new_init(self, *args, **kwargs):
128 """The modified Processor init"""
129 old_init(self, *args, looper=loop_type, **kwargs)
131 cls.__init__ = new_init
133 return cls
135 return dec
138single_loop = execution_workflow(LoopType.SingleLoop)
139"""A decorator shortcut to define a single execution processor."""
141for_loop = execution_workflow(LoopType.ForLoop)
142"""A decorator shortcut to define a for loop execution processor."""
144while_loop = execution_workflow(LoopType.WhileLoop)
145"""A decorator shortcut to define a while loop execution processor."""
148def depends_on_optional(
149 module_name: str, raise_ex: bool = False, warn: bool = True
150) -> Callable[[F], Callable[..., Any]]:
151 """
152 Function decorator to check if module_name is available.
154 If module_name is found, then returns the wrapped function. If not, its behavior depends on the raise_ex and
155 warn_only values. If raise_ex is True, then an ImportError exception is raised. If it is False and warn is
156 True, then a warning message is displayed but no exception is raised. If they are both False, then function is
157 silently skipped.
159 If raise_ex is True, the value of `warn` is not taken into account.
161 **Typical usage**
163 The user should decorate functions or class methods when they cannot be executed without the optional library.
164 In the specific case of Processor subclass, where the class itself can be created also without the missing
165 library, but it is required somewhere in the processor execution, then the user is suggested to decorate the
166 execute method with this decorator.
168 :param module_name: The optional module(s) from which the function depends on. A ";" separated list of modules can
169 also be provided.
170 :type module_name: str
171 :param raise_ex: Flag to raise an exception if module_name is not found, defaults to False.
172 :type raise_ex: bool, Optional
173 :param warn: Flag to display a warning message if module_name is not found, default to True.
174 :type warn: bool, Optional
175 :return: The wrapped function
176 :rtype: Callable
177 :raise ImportError: if module_name is not found and raise_ex is True.
178 """
180 def decorator(func: F) -> Callable[..., Any]:
181 @functools.wraps(func)
182 def wrapper(*args: Any, **kwargs: Any) -> Any:
183 all_mods_found = all(find_spec(mod.strip()) is not None for mod in module_name.split(';'))
184 if not all_mods_found:
185 msg = f'Optional dependency {module_name} not found ({func.__qualname__})'
186 if raise_ex:
187 raise ImportError(msg)
188 else:
189 if warn:
190 warnings.warn(MissingOptionalDependency(msg), stacklevel=2)
191 return None # Explicitly return None when skipping the function
192 else:
193 return func(*args, **kwargs)
195 return wrapper
197 return decorator
200def processor_depends_on_optional(
201 module_name: str, raise_ex: bool = False, warn: bool = True
202) -> Callable[[Type[P]], Type[Processor]]:
203 """
204 Class decorator factory to check if module module_name is available.
206 It checks if all the optional modules listed in `module_name` separated by a ';' can be found.
208 If all modules are found, then the class is returned as it is.
210 If at least one module is not found:
211 - and raise_ex is True, an ImportError exception is raised and the user is responsible to deal with it.
212 - if raise_ex is False, instead of returning the class, the :class:`~.Processor` is returned.
213 - depending on the value of warn, the user will be informed with a warning message or not.
215 **Typical usage**
217 The user should decorate Processor subclasses everytime the optional module is required in their __init__ method.
218 Should the check on the optional module have a positive outcome, then the Processor subclass is returned.
219 Otherwise, if raise_ex is False, an instance of the base :py:class:`~.Processor` is returned. In
220 this way, the returned class can still be executed without breaking the execution scheme but of course, without
221 producing any output.
223 Should be possible to run the __init__ method of the class without the missing library, then the user can also
224 follow the approach described in this other :func:`example <depends_on_optional>`.
226 :param module_name: The optional module(s) from which the class depends on. A ";" separated list of modules can
227 also be provided.
228 :type module_name: str
229 :param raise_ex: Flag to raise an exception if module_name not found, defaults to False.
230 :type raise_ex: bool, Optional
231 :param warn: Flag to display a warning message if module_name is not found, defaults to True.
232 :type warn: bool, Optional
233 :return: The wrapped processor.
234 :rtype: type(Processor)
235 :raise ImportError: if module_name is not found and raise_ex is True.
236 """
238 def decorator(cls: Type[P]) -> Type[Processor]:
239 """
240 The class decorator.
242 It checks if all the modules provided by the decorator factory are available on the systems.
243 If yes, then it simply returns `cls`. If no, it returns a subclass of the :class:`~.Processor`
244 after all the introspection properties have been taken from `cls`.
246 :param cls: The class being decorated.
247 :type cls: type(Processor)
248 :return: The decorated class, either cls or a subclass of :class:`~autorad.processor.Processor`.
249 :rtype: type(Processor)
250 """
252 def class_wrapper(klass: Type[Processor]) -> Type[Processor]:
253 """
254 Copy introspection properties from cls to klass.
256 :param klass: The class to be modified.
257 :type klass: class.
258 :return: The modified class.
259 :rtype: class.
260 """
261 klass.__module__ = cls.__module__
262 klass.__name__ = f'{cls.__name__} (Missing {module_name})'
263 klass.__qualname__ = cls.__qualname__
264 klass.__annotations__ = cls.__annotations__
265 klass.__doc__ = cls.__doc__
266 return klass
268 all_mods_found = all([find_spec(mod.strip()) is not None for mod in module_name.split(';')])
269 if not all_mods_found:
270 msg = f'Optional dependency {module_name} not found ({cls.__qualname__})'
271 if raise_ex:
272 raise ImportError(msg)
273 else:
274 if warn:
275 warnings.warn(MissingOptionalDependency(msg), stacklevel=2)
277 # We subclass the basic processor.
278 class NewClass(Processor):
279 pass
281 # The class wrapper is copying introspection properties from the cls to the NewClass
282 new_class = class_wrapper(NewClass)
284 else:
285 new_class = cls
286 return new_class
288 return decorator
291def class_depends_on_optional(
292 module_name: str, raise_ex: bool = False, warn: bool = True
293) -> Callable[[Type[Any]], Type[Any]]:
294 """
295 Class decorator factory to check if module module_name is available.
297 It checks if all the optional modules listed in `module_name` separated by a ';' can be found.
299 If all modules are found, then the class is returned as it is.
301 If at least one module is not found:
302 - and raise_ex is True, an ImportError exception is raised and the user is responsible to deal with it.
303 - if raise_ex is False, instead of returning the class, a new empty class is returned.
304 - depending on the value of warn, the user will be informed with a warning message or not.
306 :param module_name: The optional module(s) from which the class depends on. A ";" separated list of modules can
307 also be provided.
308 :type module_name: str
309 :param raise_ex: Flag to raise an exception if module_name not found, defaults to False.
310 :type raise_ex: bool, Optional
311 :param warn: Flag to display a warning message if module_name is not found, defaults to True.
312 :type warn: bool, Optional
313 :return: The wrapped class.
314 :rtype: type(object)
315 :raise ImportError: if module_name is not found and raise_ex is True.
316 """
318 def decorator(cls: Type[Any]) -> Type[Any]:
319 """
320 The class decorator.
322 It checks if all the modules provided by the decorator factory are available on the systems.
323 If yes, then it simply returns `cls`. If no, it returns a subclass of the cls bases.
324 after all the introspection properties have been taken from `cls`.
326 :param cls: The class being decorated.
327 :type cls: type(cls)
328 :return: The decorated class, either cls or a subclass of cls.
329 :rtype: type(cls)
330 """
332 def class_wrapper(klass: Type[Any]) -> Type[Any]:
333 """
334 Copy introspection properties from cls to klass.
336 :param klass: The class to be modified.
337 :type klass: class.
338 :return: The modified class.
339 :rtype: class.
340 """
341 klass.__module__ = cls.__module__
342 klass.__name__ = f'{cls.__name__} (Missing {module_name})'
343 klass.__qualname__ = cls.__qualname__
344 klass.__annotations__ = cls.__annotations__
345 klass.__doc__ = cls.__doc__
346 return klass
348 all_mods_found = all([find_spec(mod.strip()) is not None for mod in module_name.split(';')])
349 if not all_mods_found:
350 msg = f'Optional dependency {module_name} not found ({cls.__qualname__})'
351 if raise_ex:
352 raise ImportError(msg)
353 else:
354 if warn:
355 warnings.warn(MissingOptionalDependency(msg), stacklevel=2)
357 # we subclass the original class.
358 class NewClass(*cls.__bases__): # type: ignore
359 pass
361 # the class wrapper is copying introspection properties from the cls to the NewClass
362 new_class = class_wrapper(NewClass)
364 else:
365 new_class = cls
366 return new_class
368 return decorator