Coverage for src / mafw / lazy_import.py: 96%
70 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"""
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"""
19from __future__ import annotations
21import importlib
22import threading
23from abc import ABC, abstractmethod
24from typing import Any, Generic, Protocol, Type, TypeVar, cast, runtime_checkable
26from mafw.models import ParameterSchema
27from mafw.models.filter_schema import FilterSchema
28from mafw.models.processor_schema import ProcessorSchema
29from mafw.processor import Processor
30from mafw.ui.abstract_user_interface import UserInterfaceBase
33@runtime_checkable
34class ProcessorClassProtocol(Protocol):
35 """
36 Protocol for Processor classes.
38 .. versionadded:: v2.0.0
40 :cvar plugin_name: The name of the plugin.
41 :cvar plugin_qualname: The qualified name of the plugin.
42 :cvar __name__: The name of the class.
43 """
45 plugin_name: str
46 plugin_qualname: str
47 __name__: str
49 def parameter_schema(self) -> list[ParameterSchema]:
50 """Return the parameter schema."""
51 ...
53 def filter_schema(self) -> FilterSchema | None:
54 """Return the filter schema."""
55 ...
57 def processor_schema(self) -> ProcessorSchema:
58 """Return the processor schema."""
59 ...
61 def __call__(self, *args: Any, **kwargs: Any) -> Processor:
62 """
63 Instantiate the Processor class.
65 :param args: Positional arguments for the Processor constructor.
66 :param kwargs: Keyword arguments for the Processor constructor.
67 :return: An instance of Processor.
68 :rtype: Processor
69 """
70 ... # pragma: no cov
73@runtime_checkable
74class UserInterfaceClassProtocol(Protocol):
75 """
76 Protocol for UserInterface classes.
78 .. versionadded:: v2.0.0
80 :cvar plugin_name: The name of the plugin.
81 :cvar plugin_qualname: The qualified name of the plugin.
82 :cvar name: The name of the user interface.
83 """
85 plugin_name: str
86 plugin_qualname: str
87 name: str
89 def __call__(self, *args: Any, **kwargs: Any) -> UserInterfaceBase:
90 """
91 Instantiate the UserInterfaceBase class.
93 :param args: Positional arguments for the UserInterfaceBase constructor.
94 :param kwargs: Keyword arguments for the UserInterfaceBase constructor.
95 :return: An instance of UserInterfaceBase.
96 :rtype: UserInterfaceBase
97 """
98 ... # pragma: no cov
101T = TypeVar('T') # Class type (e.g., Processor class)
102"""The class type to be used for the generic lazy import plugin."""
103R = TypeVar('R') # Instance type (e.g., Processor instance)
104"""The instance type to be used for the generic lazy import plugin."""
107class LazyImportPlugin(Generic[T, R], ABC):
108 """
109 Proxy object that lazily imports a plugin class only when accessed.
110 Thread-safe and transparent to the user.
112 .. versionadded:: v2.0.0
113 """
115 def __init__(self, module: str, class_name: str) -> None:
116 """
117 Constructor parameter:
119 :param module: The module name where the class is located.
120 :type module: str
121 :param class_name: The name of the class to be lazily imported.
122 :type class_name: str
123 """
124 self._module = module
125 self._class_name = class_name
126 self._cached: Type[T] | None = None
127 self._lock = threading.Lock()
129 self.plugin_name = class_name
130 self.plugin_qualname = f'{module}.{class_name}'
132 @abstractmethod
133 def _post_load(self, cls: Type[T]) -> Type[T]:
134 """
135 Perform operations after loading the class.
137 :param cls: The class type that has been loaded.
138 :type cls: Type[T]
139 :return: The class type after post-load operations.
140 :rtype: Type[T]
141 """
142 return cls # pragma: no cov
144 def _load(self) -> Type[T]:
145 """
146 Load the class from the specified module.
148 :return: The loaded class type.
149 :rtype: Type[T]
150 """
151 if self._cached is None:
152 with self._lock:
153 module = importlib.import_module(self._module)
154 cls = getattr(module, self._class_name)
155 self._cached = self._post_load(cls)
156 return self._cached
158 # Allow calling class attributes transparently
159 def __getattr__(self, item: str) -> Any:
160 """
161 Access attributes of the lazily loaded class.
163 :param item: The attribute name to access.
164 :type item: str
165 :return: The attribute value.
166 :rtype: Any
167 """
168 return getattr(self._load(), item)
170 # Allow instantiation: LazyPlugin() behaves like ActualClass()
171 def __call__(self, *args: Any, **kwargs: Any) -> R:
172 """
173 Instantiate the lazily loaded class.
175 :param args: Positional arguments for the class constructor.
176 :param kwargs: Keyword arguments for the class constructor.
177 :return: An instance of the class.
178 :rtype: R
179 """
180 cls = self._load()
181 return cast(R, cls(*args, **kwargs))
183 def __repr__(self) -> str:
184 """
185 Return a string representation of the LazyImportPlugin.
187 :return: The string representation.
188 :rtype: str
189 """
190 return f'LazyImportPlugin("{self._module}", "{self._class_name}")' # pragma: no cov
193class LazyImportProcessor(LazyImportPlugin[Processor, Processor]):
194 """
195 Lazy import proxy for Processor classes.
197 .. versionadded:: v2.0.0
198 """
200 def _post_load(self, cls: Type[Processor]) -> Type[Processor]:
201 """
202 Perform operations after loading the Processor class.
204 :param cls: The Processor class type that has been loaded.
205 :type cls: Type[Processor]
206 :return: The Processor class type after post-load operations.
207 :rtype: Type[Processor]
208 """
209 return cls
211 def __repr__(self) -> str:
212 """
213 Return a string representation of the LazyImportProcessor.
215 :return: The string representation.
216 :rtype: str
217 """
218 return f'LazyImportProcessor("{self._module}", "{self._class_name}")'
221class LazyImportUserInterface(LazyImportPlugin[UserInterfaceBase, UserInterfaceBase]):
222 """
223 Lazy import proxy for UserInterface classes.
225 .. versionadded:: v2.0.0
226 """
228 def __init__(self, module: str, class_name: str, ui_name: str) -> None:
229 """
230 Constructor parameter:
232 :param module: The module name where the class is located.
233 :type module: str
234 :param class_name: The name of the class to be lazily imported.
235 :type class_name: str
236 :param ui_name: The expected name of the user interface.
237 :type ui_name: str
238 """
239 super().__init__(module, class_name)
240 self.name = ui_name
242 def _post_load(self, cls: Type[UserInterfaceBase]) -> Type[UserInterfaceBase]:
243 """
244 Perform operations after loading the UserInterface class.
246 :param cls: The UserInterface class type that has been loaded.
247 :type cls: Type[UserInterfaceBase]
248 :return: The UserInterface class type after post-load operations.
249 :rtype: Type[UserInterfaceBase]
250 :raises ValueError: If the class name is inconsistent with the expected name.
251 """
252 if getattr(cls, 'name', None) != self.name:
253 raise ValueError(f'UserInterface class {cls} has inconsistent .name: expected {self.name}')
254 return cls
256 def __repr__(self) -> str:
257 """
258 Return a string representation of the LazyImportUserInterface.
260 :return: The string representation.
261 :rtype: str
262 """
263 return f'LazyImportUserInterface("{self._module}", "{self._class_name}", "{self.name}")'