Coverage for src / mafw / lazy_import.py: 100%
60 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"""
5This module provides classes for lazy importing of plugins, ensuring thread-safe access and transparent usage.
7Classes:
8 - LazyImportPlugin: A generic class for lazy importing of plugin classes.
9 - LazyImportProcessor: A specialised class for lazy importing of Processor plugins.
10 - LazyImportUserInterface: A specialised class for lazy importing of UserInterface plugins.
12Protocols:
13 - ProcessorClassProtocol: Defines the expected interface for Processor classes.
14 - UserInterfaceClassProtocol: Defines the expected interface for UserInterface classes.
16.. versionadded:: v2.0.0
17"""
19import importlib
20import threading
21from abc import ABC, abstractmethod
22from typing import Any, Generic, Protocol, Type, TypeVar, cast, runtime_checkable
24from mafw.processor import Processor
25from mafw.ui.abstract_user_interface import UserInterfaceBase
28@runtime_checkable
29class ProcessorClassProtocol(Protocol):
30 """
31 Protocol for Processor classes.
33 .. versionadded:: v2.0.0
35 :cvar plugin_name: The name of the plugin.
36 :cvar plugin_qualname: The qualified name of the plugin.
37 :cvar __name__: The name of the class.
38 """
40 plugin_name: str
41 plugin_qualname: str
42 __name__: str
44 def __call__(self, *args: Any, **kwargs: Any) -> Processor:
45 """
46 Instantiate the Processor class.
48 :param args: Positional arguments for the Processor constructor.
49 :param kwargs: Keyword arguments for the Processor constructor.
50 :return: An instance of Processor.
51 :rtype: Processor
52 """
53 ... # pragma: no cov
56@runtime_checkable
57class UserInterfaceClassProtocol(Protocol):
58 """
59 Protocol for UserInterface classes.
61 .. versionadded:: v2.0.0
63 :cvar plugin_name: The name of the plugin.
64 :cvar plugin_qualname: The qualified name of the plugin.
65 :cvar name: The name of the user interface.
66 """
68 plugin_name: str
69 plugin_qualname: str
70 name: str
72 def __call__(self, *args: Any, **kwargs: Any) -> UserInterfaceBase:
73 """
74 Instantiate the UserInterfaceBase class.
76 :param args: Positional arguments for the UserInterfaceBase constructor.
77 :param kwargs: Keyword arguments for the UserInterfaceBase constructor.
78 :return: An instance of UserInterfaceBase.
79 :rtype: UserInterfaceBase
80 """
81 ... # pragma: no cov
84T = TypeVar('T') # Class type (e.g., Processor class)
85"""The class type to be used for the generic lazy import plugin."""
86R = TypeVar('R') # Instance type (e.g., Processor instance)
87"""The instance type to be used for the generic lazy import plugin."""
90class LazyImportPlugin(Generic[T, R], ABC):
91 """
92 Proxy object that lazily imports a plugin class only when accessed.
93 Thread-safe and transparent to the user.
95 .. versionadded:: v2.0.0
96 """
98 def __init__(self, module: str, class_name: str) -> None:
99 """
100 Constructor parameter:
102 :param module: The module name where the class is located.
103 :type module: str
104 :param class_name: The name of the class to be lazily imported.
105 :type class_name: str
106 """
107 self._module = module
108 self._class_name = class_name
109 self._cached: Type[T] | None = None
110 self._lock = threading.Lock()
112 self.plugin_name = class_name
113 self.plugin_qualname = f'{module}.{class_name}'
115 @abstractmethod
116 def _post_load(self, cls: Type[T]) -> Type[T]:
117 """
118 Perform operations after loading the class.
120 :param cls: The class type that has been loaded.
121 :type cls: Type[T]
122 :return: The class type after post-load operations.
123 :rtype: Type[T]
124 """
125 return cls # pragma: no cov
127 def _load(self) -> Type[T]:
128 """
129 Load the class from the specified module.
131 :return: The loaded class type.
132 :rtype: Type[T]
133 """
134 if self._cached is None:
135 with self._lock:
136 module = importlib.import_module(self._module)
137 cls = getattr(module, self._class_name)
138 self._cached = self._post_load(cls)
139 return self._cached
141 # Allow calling class attributes transparently
142 def __getattr__(self, item: str) -> Any:
143 """
144 Access attributes of the lazily loaded class.
146 :param item: The attribute name to access.
147 :type item: str
148 :return: The attribute value.
149 :rtype: Any
150 """
151 return getattr(self._load(), item)
153 # Allow instantiation: LazyPlugin() behaves like ActualClass()
154 def __call__(self, *args: Any, **kwargs: Any) -> R:
155 """
156 Instantiate the lazily loaded class.
158 :param args: Positional arguments for the class constructor.
159 :param kwargs: Keyword arguments for the class constructor.
160 :return: An instance of the class.
161 :rtype: R
162 """
163 cls = self._load()
164 return cast(R, cls(*args, **kwargs))
166 def __repr__(self) -> str:
167 """
168 Return a string representation of the LazyImportPlugin.
170 :return: The string representation.
171 :rtype: str
172 """
173 return f'LazyImportPlugin("{self._module}", "{self._class_name}")' # pragma: no cov
176class LazyImportProcessor(LazyImportPlugin[Processor, Processor]):
177 """
178 Lazy import proxy for Processor classes.
180 .. versionadded:: v2.0.0
181 """
183 def _post_load(self, cls: Type[Processor]) -> Type[Processor]:
184 """
185 Perform operations after loading the Processor class.
187 :param cls: The Processor class type that has been loaded.
188 :type cls: Type[Processor]
189 :return: The Processor class type after post-load operations.
190 :rtype: Type[Processor]
191 """
192 return cls
194 def __repr__(self) -> str:
195 """
196 Return a string representation of the LazyImportProcessor.
198 :return: The string representation.
199 :rtype: str
200 """
201 return f'LazyImportProcessor("{self._module}", "{self._class_name}")'
204class LazyImportUserInterface(LazyImportPlugin[UserInterfaceBase, UserInterfaceBase]):
205 """
206 Lazy import proxy for UserInterface classes.
208 .. versionadded:: v2.0.0
209 """
211 def __init__(self, module: str, class_name: str, ui_name: str) -> None:
212 """
213 Constructor parameter:
215 :param module: The module name where the class is located.
216 :type module: str
217 :param class_name: The name of the class to be lazily imported.
218 :type class_name: str
219 :param ui_name: The expected name of the user interface.
220 :type ui_name: str
221 """
222 super().__init__(module, class_name)
223 self.name = ui_name
225 def _post_load(self, cls: Type[UserInterfaceBase]) -> Type[UserInterfaceBase]:
226 """
227 Perform operations after loading the UserInterface class.
229 :param cls: The UserInterface class type that has been loaded.
230 :type cls: Type[UserInterfaceBase]
231 :return: The UserInterface class type after post-load operations.
232 :rtype: Type[UserInterfaceBase]
233 :raises ValueError: If the class name is inconsistent with the expected name.
234 """
235 if getattr(cls, 'name', None) != self.name:
236 raise ValueError(f'UserInterface class {cls} has inconsistent .name: expected {self.name}')
237 return cls
239 def __repr__(self) -> str:
240 """
241 Return a string representation of the LazyImportUserInterface.
243 :return: The string representation.
244 :rtype: str
245 """
246 return f'LazyImportUserInterface("{self._module}", "{self._class_name}", "{self.name}")'