Coverage for src/metaclass_registry/discovery.py: 83%
100 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-02 00:58 +0000
1"""
2Generic registry class discovery utility.
4Consolidates duplicated registry discovery patterns across:
5- Library registries (processing backends)
6- Format registries (experimental analysis)
7- Microscope handler registries
8- Storage backend registries
10This module eliminates ~70 lines of duplicated pkgutil + importlib boilerplate
11by providing a single, well-tested discovery function.
12"""
14import importlib
15import inspect
16import logging
17import pkgutil
18import threading
19from collections.abc import Callable, Iterable
20from types import ModuleType
22logger = logging.getLogger(__name__)
23_ROOT_LOGGING_IMPORT_LOCK = threading.RLock()
26def import_module_preserving_root_logging(
27 module_name: str,
28 package: str | None = None,
29) -> ModuleType:
30 """Import a module without transferring root-logger ownership to it.
32 Plugin and callable discovery imports code owned by third parties. Some
33 packages configure the process root logger at import time, which can replace
34 application handlers or write diagnostics into machine-readable stdout.
35 Discovery owns importing those modules, but the hosting application retains
36 ownership of root logging.
38 A temporary handler prevents ordinary ``logging.basicConfig`` calls from
39 installing handlers. The prior root handler set, level, disabled state, and
40 global logging threshold are restored even when the import fails. Named
41 package loggers remain under the imported package's control.
42 """
44 with _ROOT_LOGGING_IMPORT_LOCK:
45 root_logger = logging.getLogger()
46 previous_handlers = tuple(root_logger.handlers)
47 previous_level = root_logger.level
48 previous_disabled = root_logger.disabled
49 previous_disable_threshold = root_logger.manager.disable
50 guard_handler = logging.NullHandler()
51 root_logger.handlers = [guard_handler]
52 try:
53 return importlib.import_module(module_name, package)
54 finally:
55 temporary_handlers = tuple(root_logger.handlers)
56 root_logger.handlers = list(previous_handlers)
57 root_logger.setLevel(previous_level)
58 root_logger.disabled = previous_disabled
59 logging.disable(previous_disable_threshold)
60 previous_handler_ids = {id(handler) for handler in previous_handlers}
61 for handler in temporary_handlers:
62 if id(handler) not in previous_handler_ids:
63 handler.close()
66def discover_registry_classes(
67 package_path: Iterable[str],
68 package_prefix: str,
69 base_class: type,
70 exclude_modules: set[str] | None = None,
71 validation_func: Callable[[type], bool] | None = None,
72 skip_packages: bool = True,
73 module_filter: Callable[[str], bool] | None = None,
74) -> list[type]:
75 """
76 Generic registry class discovery using pkgutil + importlib pattern.
78 Scans a package for classes that inherit from a base class and automatically
79 discovers them for registration. This eliminates duplicated discovery code
80 across different registry systems.
82 Args:
83 package_path: Package __path__ attribute to scan (e.g., openhcs.io.__path__)
84 Accepts any iterable of strings (List, Tuple, _NamespacePath, etc.)
85 package_prefix: Module prefix for importlib (e.g., "openhcs.io.")
86 base_class: Base class to filter for (e.g., StorageBackend)
87 exclude_modules: Set of module name substrings to skip (e.g., {'base', 'registry'})
88 validation_func: Optional function to validate discovered classes
89 Should return True to include, False to exclude
90 skip_packages: If True, skip package directories (default: True)
91 module_filter: Declaration-owned eligibility before import. Selected
92 import failures propagate instead of incomplete absence.
94 Returns:
95 List of discovered registry classes
97 Example:
98 >>> from openhcs.io.base import StorageBackend
99 >>> import openhcs.io
100 >>> backends = discover_registry_classes(
101 ... package_path=openhcs.io.__path__,
102 ... package_prefix="openhcs.io.",
103 ... base_class=StorageBackend,
104 ... exclude_modules={'base', 'backend_registry'}
105 ... )
106 >>> print([b.__name__ for b in backends])
107 ['DiskStorageBackend', 'MemoryStorageBackend', 'ZarrStorageBackend']
108 """
109 registry_classes = []
110 exclude_modules = exclude_modules or set()
112 logger.debug(
113 f"Discovering registry classes: base={base_class.__name__}, "
114 f"prefix={package_prefix}, exclude={exclude_modules}"
115 )
117 for importer, module_name, ispkg in pkgutil.iter_modules(package_path, package_prefix):
118 # Skip packages if requested
119 if ispkg and skip_packages:
120 continue
122 # Skip excluded modules
123 if any(excluded in module_name for excluded in exclude_modules):
124 logger.debug(f"Skipping excluded module: {module_name}")
125 continue
127 if module_filter is not None and not module_filter(module_name):
128 continue
130 try:
131 # Import the module
132 module = import_module_preserving_root_logging(module_name)
134 # Find all classes in the module
135 for name, obj in inspect.getmembers(module, inspect.isclass):
136 # Filter for subclasses of base_class
137 if not issubclass(obj, base_class):
138 continue
140 # Exclude the base class itself
141 if obj is base_class:
142 continue
144 # Only include classes defined in this module (not imported)
145 if obj.__module__ != module_name:
146 continue
148 # Apply optional validation function
149 if validation_func and not validation_func(obj):
150 logger.debug(f"Validation failed for {obj.__name__}")
151 continue
153 logger.debug(f"Discovered registry class: {obj.__name__} from {module_name}")
154 registry_classes.append(obj)
156 except ImportError as e:
157 if module_filter is not None:
158 raise
159 # Skip modules that can't be imported (e.g., missing optional dependencies)
160 logger.debug(f"Could not import module {module_name}: {e}")
161 continue
162 except Exception as e:
163 if module_filter is not None:
164 raise
165 # Log unexpected errors but continue discovery
166 logger.warning(f"Failed to load registry module {module_name}: {e}")
167 continue
169 logger.info(
170 f"Discovered {len(registry_classes)} registry classes for {base_class.__name__}: "
171 f"{[cls.__name__ for cls in registry_classes]}"
172 )
174 return registry_classes
177def discover_registry_classes_recursive(
178 package_path: Iterable[str],
179 package_prefix: str,
180 base_class: type,
181 exclude_modules: set[str] | None = None,
182 validation_func: Callable[[type], bool] | None = None,
183) -> list[type]:
184 """
185 Recursive version of discover_registry_classes that walks entire package tree.
187 Uses pkgutil.walk_packages instead of iter_modules to recursively scan
188 all subpackages. Useful for deeply nested registry structures.
190 Args:
191 package_path: Package __path__ attribute to scan
192 Accepts any iterable of strings (List, Tuple, _NamespacePath, etc.)
193 package_prefix: Module prefix for importlib
194 base_class: Base class to filter for
195 exclude_modules: Set of module name substrings to skip
196 validation_func: Optional function to validate discovered classes
198 Returns:
199 List of discovered registry classes
201 Example:
202 >>> from openhcs.processing.backends.lib_registry.unified_registry import (
203 ... LibraryRegistryBase,
204 ... )
205 >>> import openhcs.processing.backends.experimental_analysis
206 >>> registries = discover_registry_classes_recursive(
207 ... package_path=openhcs.processing.backends.experimental_analysis.__path__,
208 ... package_prefix="openhcs.processing.backends.experimental_analysis.",
209 ... base_class=MicroscopeFormatRegistryBase,
210 ... exclude_modules={'base'}
211 ... )
212 """
213 registry_classes = []
214 exclude_modules = exclude_modules or set()
216 logger.debug(
217 f"Discovering registry classes (recursive): base={base_class.__name__}, "
218 f"prefix={package_prefix}, exclude={exclude_modules}"
219 )
221 # Walk through all modules in the package tree
222 for importer, modname, ispkg in pkgutil.walk_packages(package_path, prefix=package_prefix):
223 # Skip packages (only process modules)
224 if ispkg:
225 continue
227 # Skip excluded modules
228 if any(excluded in modname for excluded in exclude_modules):
229 logger.debug(f"Skipping excluded module: {modname}")
230 continue
232 try:
233 # Import the module
234 module = import_module_preserving_root_logging(modname)
236 # Find all classes in the module
237 for attr_name in dir(module):
238 attr = getattr(module, attr_name)
240 # Check if it's a class
241 if not isinstance(attr, type):
242 continue
244 # Check if it's a subclass of base_class
245 if not issubclass(attr, base_class):
246 continue
248 # Exclude the base class itself
249 if attr is base_class:
250 continue
252 # Apply optional validation function
253 if validation_func and not validation_func(attr):
254 logger.debug(f"Validation failed for {attr.__name__}")
255 continue
257 logger.debug(f"Discovered registry class: {attr.__name__} from {modname}")
258 registry_classes.append(attr)
260 except ImportError as e:
261 # Skip modules that can't be imported
262 logger.debug(f"Could not import module {modname}: {e}")
263 continue
264 except Exception as e:
265 # Log unexpected errors but continue discovery
266 logger.warning(f"Failed to load registry module {modname}: {e}")
267 continue
269 logger.info(
270 "Discovered %d registry classes (recursive) for %s: %s",
271 len(registry_classes),
272 base_class.__name__,
273 [cls.__name__ for cls in registry_classes],
274 )
276 return registry_classes