Coverage for src/metaclass_registry/core.py: 77%
394 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-02 00:58 +0000
1"""
2Generic metaclass infrastructure for automatic plugin registration.
4This module provides reusable metaclass infrastructure for Pattern A registry systems
5(1:1 class-to-plugin mapping with automatic discovery). It eliminates code duplication
6across MicroscopeHandlerMeta, StorageBackendMeta, and ContextProviderMeta.
8Pattern Selection Guide:
9-----------------------
10Use AutoRegisterMeta (Pattern A) when:
11- You have a 1:1 mapping between classes and plugins
12- Plugins should be automatically discovered and registered
13- Registration happens at class definition time
14- Simple metadata (just a key and maybe one secondary registry)
16Use Service Pattern (Pattern B) when:
17- You have many-to-one mapping (multiple items per plugin)
18- Complex metadata (FunctionMetadata with 8+ fields)
19- Need aggregation across multiple sources
20- Examples: Function registry, Format registry
22Use Functional Registry (Pattern C) when:
23- Simple type-to-handler mappings
24- No state needed
25- Functional programming style preferred
26- Examples: Widget creation registries
28Use Manual Registration (Pattern D) when:
29- Complex initialization logic required
30- Explicit control over registration timing needed
31- Very few plugins (< 3)
32- Examples: ZMQ servers, Pipeline steps
34Architecture:
35------------
36AutoRegisterMeta uses a configuration-driven approach:
371. RegistryConfig defines registration behavior
382. The nominal registry root owns that configuration as ``__registry_config__``
393. AutoRegisterMeta applies the configuration during class creation
41This keeps domain semantics on their nominal owner while eliminating duplicated
42metaclass wrappers and caller-side discovery inference.
43"""
45import importlib
46import logging
47import threading
48from abc import ABCMeta
49from collections.abc import Callable, Hashable, Iterator
50from dataclasses import dataclass
51from enum import Enum, EnumMeta
52from typing import Any
54from .cache import (
55 CacheConfig,
56 RegistryCacheManager,
57 deserialize_plugin_class,
58 get_package_file_mtimes,
59 serialize_plugin_class,
60)
62logger = logging.getLogger(__name__)
64# Type aliases for clarity
65RegistryKey = Hashable
66RegistryDict = dict[RegistryKey, Any]
67KeyExtractor = Callable[[str, type], RegistryKey]
69# Constants for key sources
70PRIMARY_KEY = "primary"
73class RegistryKeyAttribute(str, Enum):
74 """Common class-attribute names used as registry keys."""
76 REGISTRY_KEY = "registry_key"
77 STRATEGY_KEY = "strategy_key"
78 STRATEGY_LABEL = "strategy_label"
79 VALUE_TYPE_LABEL = "value_type_label"
80 LAYOUT_KEY = "layout_key"
83class SecondaryRegistryDict(dict[RegistryKey, Any]):
84 """
85 Dict for secondary registries that auto-triggers primary registry discovery.
87 When accessed, this dict triggers discovery of the primary registry,
88 which populates both the primary and secondary registries.
89 """
91 def __init__(self, primary_registry: "LazyDiscoveryDict") -> None:
92 super().__init__()
93 self._primary_registry = primary_registry
95 def _ensure_discovered(self) -> None:
96 """Trigger discovery of primary registry (which populates this secondary registry)."""
97 if hasattr(self._primary_registry, "_discover"):
98 self._primary_registry._discover()
100 def __getitem__(self, key: RegistryKey) -> Any:
101 self._ensure_discovered()
102 return super().__getitem__(key)
104 def __contains__(self, key: object) -> bool:
105 self._ensure_discovered()
106 return super().__contains__(key)
108 def __iter__(self) -> Iterator[RegistryKey]:
109 self._ensure_discovered()
110 return super().__iter__()
112 def __len__(self) -> int:
113 self._ensure_discovered()
114 return super().__len__()
116 def keys(self) -> Any:
117 self._ensure_discovered()
118 return super().keys()
120 def values(self) -> Any:
121 self._ensure_discovered()
122 return super().values()
124 def items(self) -> Any:
125 self._ensure_discovered()
126 return super().items()
128 def get(self, key: RegistryKey, default: Any = None) -> Any:
129 self._ensure_discovered()
130 return super().get(key, default)
133class LazyDiscoveryDict(dict[RegistryKey, Any]):
134 """
135 Dict that auto-discovers plugins on first access with optional caching.
137 Supports caching discovered plugins to speed up subsequent application starts.
138 Cache is validated against package version and file modification times.
140 Thread-safe: Uses locking to ensure discovery happens only once
141 even when accessed from multiple threads simultaneously.
142 """
144 def __init__(self, enable_cache: bool = True) -> None:
145 """
146 Initialize lazy discovery dict.
148 Args:
149 enable_cache: If True, use caching to speed up discovery
150 """
151 super().__init__()
152 self._base_class: type | None = None
153 self._config: RegistryConfig | None = None
154 self._discovered = False
155 self._enable_cache = enable_cache
156 self._cache_manager: RegistryCacheManager[type] | None = None
157 self._discovery_lock = threading.RLock() # Reentrant lock for same-thread re-entry
158 self._discovery_package: str | None = None # Store for pickling support
160 def _set_config(self, base_class: type, config: "RegistryConfig") -> None:
161 self._base_class = base_class
162 self._config = config
163 self._discovery_package = config.discovery_package # Store for pickling support
165 # Initialize cache manager if caching is enabled
166 if self._enable_cache and config.discovery_package:
167 try:
168 self._cache_manager = RegistryCacheManager(
169 cache_name=f"{config.registry_name.replace(' ', '_')}_registry",
170 version_getter=self._get_version,
171 serializer=serialize_plugin_class,
172 deserializer=deserialize_plugin_class,
173 config=CacheConfig(
174 max_age_days=7, check_mtimes=True # Validate file modifications
175 ),
176 file_mtimes_getter=self._get_discovery_file_mtimes,
177 )
178 except Exception as e:
179 logger.debug(f"Failed to initialize cache manager: {e}")
180 self._cache_manager = None
182 def _get_version(self) -> str:
183 """
184 Get version from the discovery package for cache validation.
186 Returns:
187 Version string or 'unknown' if unable to determine
188 """
189 try:
190 # Try to get version from the root package
191 if self._discovery_package:
192 root_package = self._discovery_package.split(".")[0]
193 mod = __import__(root_package)
194 return getattr(mod, "__version__", "unknown")
195 except Exception:
196 return "unknown"
197 return "unknown"
199 def _get_discovery_file_mtimes(self) -> dict[str, float]:
200 """Return the complete source inventory for this discovery package."""
202 if not self._discovery_package or self._config is None:
203 return {}
204 return get_package_file_mtimes(
205 self._discovery_package,
206 recursive=self._config.discovery_recursive,
207 )
209 def _has_discovery_configuration(self) -> bool:
210 """Own the shared configuration/root admission for both discovery scopes."""
211 if not self._config or not self._config.discovery_package:
212 return False
213 if self._base_class is None:
214 raise RuntimeError("Discovery configuration has no nominal registry root")
215 return True
217 def _discover(self) -> None:
218 """
219 Run discovery once, using cache if available.
221 Thread-safe: Discovery happens inside the lock to ensure atomicity.
222 The lock is held for the entire duration to prevent other threads
223 from reading a partially-populated registry.
225 CRITICAL: No fast path check outside the lock! All threads must acquire
226 the lock to ensure they don't read a partially-populated registry.
227 """
228 # No config = nothing to discover
229 if not self._has_discovery_configuration():
230 return
232 # ALWAYS acquire lock - no fast path to avoid race condition
233 # RLock allows same thread to re-acquire during module imports
234 with self._discovery_lock:
235 # Check if already discovered (inside lock)
236 if self._discovered:
237 return
239 # Mark as discovered to prevent infinite re-entry from same thread
240 # (module imports during discovery might access registry)
241 self._discovered = True
243 # Try to load from cache first
244 if self._cache_manager:
245 try:
246 cached_plugins = self._cache_manager.load_cache()
247 if cached_plugins is not None:
248 # Reconstruct registry from cache
249 self.update(cached_plugins)
250 logger.debug(
251 f"✅ Loaded {len(self)} {self._config.registry_name}s from cache"
252 )
253 return
254 except Exception as e:
255 logger.debug(f"Cache load failed for {self._config.registry_name}: {e}")
257 # Cache miss or disabled - perform full discovery
258 try:
259 pkg = importlib.import_module(self._config.discovery_package)
261 if self._config.discovery_function:
262 self._config.discovery_function(
263 pkg.__path__, f"{self._config.discovery_package}.", self._base_class
264 )
265 else:
266 from metaclass_registry.discovery import (
267 discover_registry_classes,
268 discover_registry_classes_recursive,
269 )
271 func = (
272 discover_registry_classes_recursive
273 if self._config.discovery_recursive
274 else discover_registry_classes
275 )
276 func(pkg.__path__, f"{self._config.discovery_package}.", self._base_class)
278 logger.debug(f"Discovered {len(self)} {self._config.registry_name}s")
280 # Save to cache if enabled
281 if self._cache_manager:
282 try:
283 file_mtimes = self._get_discovery_file_mtimes()
284 self._cache_manager.save_cache(dict(self), file_mtimes)
285 except Exception as e:
286 logger.debug(f"Failed to save cache for {self._config.registry_name}: {e}")
288 except Exception as e:
289 logger.warning(f"Discovery failed: {e}")
290 # Lock released here - registry is now fully populated and safe to read
292 def discover_matching(self, module_filter: Callable[[str], bool]) -> None:
293 """Admit selected modules through the original discovery/import owner.
295 Selection is not full discovery and never publishes a complete cache.
296 Domain declarations supply eligibility; their ordinary metaclass still
297 owns registration. Selected import errors propagate rather than turning
298 an incomplete selection into apparent absence.
299 """
300 from .discovery import discover_registry_classes
302 if not self._has_discovery_configuration():
303 return
304 if self._config.discovery_recursive or self._config.discovery_function:
305 raise ValueError("Selected discovery requires the flat discovery owner")
306 with self._discovery_lock:
307 if self._discovered:
308 return
309 package = importlib.import_module(self._config.discovery_package)
310 discover_registry_classes(
311 package.__path__,
312 f"{self._config.discovery_package}.",
313 self._base_class,
314 module_filter=module_filter,
315 )
317 def __getitem__(self, key: RegistryKey) -> Any:
318 with self._discovery_lock:
319 self._discover()
320 return super().__getitem__(key)
322 def __contains__(self, key: object) -> bool:
323 with self._discovery_lock:
324 self._discover()
325 return super().__contains__(key)
327 def __iter__(self) -> Iterator[RegistryKey]:
328 with self._discovery_lock:
329 self._discover()
330 return super().__iter__()
332 def __len__(self) -> int:
333 with self._discovery_lock:
334 self._discover()
335 return super().__len__()
337 def keys(self) -> Any:
338 with self._discovery_lock:
339 self._discover()
340 return super().keys()
342 def values(self) -> Any:
343 with self._discovery_lock:
344 self._discover()
345 return super().values()
347 def items(self) -> Any:
348 with self._discovery_lock:
349 self._discover()
350 return super().items()
352 def get(self, key: RegistryKey, default: Any = None) -> Any:
353 with self._discovery_lock:
354 self._discover()
355 return super().get(key, default)
357 def __getstate__(self) -> dict[RegistryKey, Any]:
358 """
359 Return state for pickling, excluding non-picklable objects.
361 The _discovery_lock and _cache_manager are excluded since they contain
362 non-picklable objects (RLock and potentially closures).
363 """
364 state = dict(self) # Copy the dict contents
365 state.update(
366 {
367 "_base_class": self._base_class,
368 "_config": self._config,
369 "_discovered": self._discovered,
370 "_enable_cache": self._enable_cache,
371 "_discovery_package": self._discovery_package,
372 # Exclude: _discovery_lock, _cache_manager
373 }
374 )
375 return state
377 def __setstate__(self, state: dict[RegistryKey, Any]) -> None:
378 """
379 Restore state from pickle, recreating non-picklable objects.
381 The _discovery_lock is recreated as a new RLock.
382 The _cache_manager is set to None and will be reinitialized if needed.
383 """
384 # Restore dict contents
385 for key, value in state.items():
386 if isinstance(key, str) and key.startswith("_"):
387 setattr(self, key, value)
388 else:
389 self[key] = value
390 # Recreate non-picklable objects
391 self._discovery_lock = threading.RLock()
392 self._cache_manager = None # Will be reinitialized if needed
395@dataclass(frozen=True)
396class SecondaryRegistry:
397 """Configuration for a secondary registry (e.g., metadata handlers)."""
399 registry_dict: RegistryDict
400 key_source: str # 'primary' or attribute name
401 attr_name: str # Attribute to check on the class
404@dataclass(frozen=True)
405class RegistryFamily:
406 """Declarative registry-family configuration for ``AutoRegisterMeta`` roots.
408 ``AutoRegisterMeta`` historically used low-level class attributes such as
409 ``__registry_key__`` and ``__skip_if_no_key__``. Those attributes remain
410 the compatibility surface, but registry roots can now declare one nominal
411 family object and let the metaclass install the legacy attributes.
412 """
414 key_attribute: str | RegistryKeyAttribute
415 skip_if_no_key: bool = True
416 registry_name: str | None = None
418 @property
419 def key_attribute_name(self) -> str:
420 """Return the string attribute name consumed by ``AutoRegisterMeta``."""
421 if isinstance(self.key_attribute, RegistryKeyAttribute):
422 return self.key_attribute.value
423 return self.key_attribute
425 def apply_to_class(self, cls: type, explicit_attrs: dict) -> None:
426 """Install legacy metaclass attributes for compatibility/introspection."""
427 if "__registry_key__" not in explicit_attrs:
428 setattr(cls, "__registry_key__", self.key_attribute_name)
429 if "__skip_if_no_key__" not in explicit_attrs:
430 setattr(cls, "__skip_if_no_key__", self.skip_if_no_key)
431 if self.registry_name is not None and "__registry_name__" not in explicit_attrs:
432 setattr(cls, "__registry_name__", self.registry_name)
435@dataclass(frozen=True)
436class RegistryConfig:
437 """
438 Configuration for automatic class registration behavior.
440 This dataclass encapsulates all the configuration needed for metaclass
441 registration, making the pattern explicit and easy to understand.
443 Attributes:
444 registry_dict: Dictionary to register classes into (e.g., MICROSCOPE_HANDLERS)
445 key_attribute: Name of class attribute containing the registration key
446 (e.g., '_microscope_type', '_backend_type', '_context_type')
447 key_extractor: Optional function to derive key from class name if key_attribute
448 is not set. Signature: (class_name: str, cls: Type) -> str
449 skip_if_no_key: If True, skip registration when key_attribute is None.
450 If False, require either key_attribute or key_extractor.
451 secondary_registries: Optional list of secondary registry configurations
452 log_registration: If True, log debug message when class is registered
453 registry_name: Human-readable name for logging (e.g., 'microscope handler')
454 discovery_package: Explicit package name to auto-discover (e.g.,
455 'openhcs.microscopes'). ``None`` keeps registration local to
456 declarations imported by the application.
457 discovery_recursive: If True, use recursive discovery (default: False)
459 Examples:
460 # Microscope handlers with name-based key extraction and secondary registry
461 RegistryConfig(
462 registry_dict=MICROSCOPE_HANDLERS,
463 key_attribute='_microscope_type',
464 key_extractor=extract_key_from_handler_suffix,
465 skip_if_no_key=False,
466 secondary_registries=[
467 SecondaryRegistry(
468 registry_dict=METADATA_HANDLERS,
469 key_source=PRIMARY_KEY,
470 attr_name='_metadata_handler_class'
471 )
472 ],
473 log_registration=True,
474 registry_name='microscope handler'
475 )
477 # Storage backends with explicit key and skip-if-none behavior
478 RegistryConfig(
479 registry_dict=STORAGE_BACKENDS,
480 key_attribute='_backend_type',
481 skip_if_no_key=True,
482 registry_name='storage backend'
483 )
485 # Context providers with simple explicit key
486 RegistryConfig(
487 registry_dict=CONTEXT_PROVIDERS,
488 key_attribute='_context_type',
489 skip_if_no_key=True,
490 registry_name='context provider'
491 )
492 """
494 registry_dict: RegistryDict
495 key_attribute: str
496 key_extractor: KeyExtractor | None = None
497 skip_if_no_key: bool = False
498 secondary_registries: list[SecondaryRegistry] | None = None
499 log_registration: bool = True
500 registry_name: str = "plugin"
501 discovery_package: str | None = None
502 discovery_recursive: bool = False
503 discovery_function: Callable[..., Any] | None = None # Custom discovery function
506class AutoRegisterMeta(ABCMeta):
507 """
508 Generic metaclass for automatic plugin registration (Pattern A).
510 This metaclass automatically registers concrete classes in a global registry
511 when they are defined, eliminating the need for manual registration calls.
513 Features:
514 - Skips abstract classes (checks __abstractmethods__)
515 - Supports explicit keys via class attributes
516 - Supports derived keys via key extraction functions
517 - Supports secondary registries (e.g., metadata handlers)
518 - Configurable skip-if-no-key behavior
519 - Debug logging for registration events
521 Usage:
522 # Create domain-specific metaclass
523 class MicroscopeHandlerMeta(AutoRegisterMeta):
524 def __new__(mcs, name, bases, attrs):
525 return super().__new__(mcs, name, bases, attrs,
526 registry_config=_MICROSCOPE_REGISTRY_CONFIG)
528 # Use in class definition
529 class ImageXpressHandler(MicroscopeHandler, metaclass=MicroscopeHandlerMeta):
530 _microscope_type = 'imagexpress' # Optional if key_extractor is provided
531 _metadata_handler_class = ImageXpressMetadata # Optional secondary registration
533 Design Principles:
534 - Explicit configuration over magic behavior
535 - Preserve all domain-specific features
536 - Zero breaking changes to existing code
537 - Easy to understand and debug
538 """
540 def __new__(
541 mcs: type["AutoRegisterMeta"],
542 name: str,
543 bases: tuple[type, ...],
544 attrs: dict[str, Any],
545 registry_config: RegistryConfig | None = None,
546 ) -> type:
547 """
548 Create a new class and register it if appropriate.
550 Args:
551 name: Name of the class being created
552 bases: Base classes
553 attrs: Class attributes dictionary
554 registry_config: Configuration for registration behavior.
555 If None, auto-configures from class attributes or skips registration.
557 Returns:
558 The newly created class
559 """
560 registry_family = attrs.get("__registry_family__")
561 declared_registry_config = attrs.get("__registry_config__")
562 if declared_registry_config is not None:
563 if not isinstance(declared_registry_config, RegistryConfig):
564 raise TypeError("__registry_config__ must be a RegistryConfig")
565 if registry_config is not None and registry_config is not declared_registry_config:
566 raise ValueError("Registry configuration must have one authoritative declaration")
567 registry_config = declared_registry_config
568 elif registry_config is None:
569 inherited_registry_configs: list[RegistryConfig] = []
570 for base in bases:
571 inherited_registry_config = getattr(
572 base,
573 "__registry_config__",
574 None,
575 )
576 if not isinstance(inherited_registry_config, RegistryConfig):
577 continue
578 if all(
579 inherited_registry_config is not existing_config
580 for existing_config in inherited_registry_configs
581 ):
582 inherited_registry_configs.append(inherited_registry_config)
583 if len(inherited_registry_configs) > 1:
584 raise TypeError(f"{name} inherits multiple registry configuration authorities")
585 registry_config = inherited_registry_configs[0] if inherited_registry_configs else None
587 # Create the class using ABCMeta
588 new_class: type = super().__new__(mcs, name, bases, attrs)
589 if isinstance(registry_family, RegistryFamily):
590 registry_family.apply_to_class(new_class, attrs)
591 if declared_registry_config is not None:
592 declared_registry = attrs.get("__registry__")
593 if (
594 "__registry__" in attrs
595 and declared_registry is not declared_registry_config.registry_dict
596 ):
597 raise ValueError(
598 f"{name}.__registry__ must be the registry owned by " "its __registry_config__"
599 )
600 setattr(new_class, "__registry__", declared_registry_config.registry_dict)
602 # Auto-configure registry if not provided but class has __registry__ attributes
603 if registry_config is None:
604 registry_config = mcs._auto_configure_registry(new_class, attrs)
605 if registry_config is None:
606 return new_class # No config and no auto-config possible
608 # Set up lazy discovery if registry dict supports it (only once for base class)
609 if (
610 isinstance(registry_config.registry_dict, LazyDiscoveryDict)
611 and not registry_config.registry_dict._config
612 ):
613 config = registry_config
615 # Auto-wrap secondary registries with SecondaryRegistryDict
616 if config.secondary_registries:
617 import sys
618 from dataclasses import replace
620 wrapped_secondaries: list[SecondaryRegistry] = []
621 module = sys.modules.get(new_class.__module__)
623 for sec_reg in config.secondary_registries:
624 # Check if secondary registry needs wrapping
625 if isinstance(sec_reg.registry_dict, dict) and not isinstance(
626 sec_reg.registry_dict, SecondaryRegistryDict
627 ):
628 # Create a new SecondaryRegistryDict wrapping the primary registry
629 wrapped_dict = SecondaryRegistryDict(registry_config.registry_dict)
630 # Copy any existing entries from the old dict
631 wrapped_dict.update(sec_reg.registry_dict)
633 # Find and update the module global variable
634 if module:
635 for var_name, var_value in vars(module).items():
636 if var_value is sec_reg.registry_dict:
637 setattr(module, var_name, wrapped_dict)
638 logger.debug(
639 "Auto-wrapped secondary registry %r in %s",
640 var_name,
641 new_class.__module__,
642 )
643 break
645 # Create new SecondaryRegistry with wrapped dict
646 wrapped_sec_reg = SecondaryRegistry(
647 registry_dict=wrapped_dict,
648 key_source=sec_reg.key_source,
649 attr_name=sec_reg.attr_name,
650 )
651 wrapped_secondaries.append(wrapped_sec_reg)
652 else:
653 wrapped_secondaries.append(sec_reg)
655 # Rebuild config with wrapped secondary registries
656 config = replace(config, secondary_registries=wrapped_secondaries)
658 registry_config.registry_dict._set_config(new_class, config)
660 # Only register concrete classes (not abstract base classes)
661 if not bases or getattr(new_class, "__abstractmethods__", None):
662 return new_class
664 # Get or derive the registration key
665 key = mcs._get_registration_key(name, new_class, registry_config)
667 # Handle missing key
668 if key is None:
669 return mcs._handle_missing_key(name, registry_config, new_class)
671 # Register in primary registry
672 mcs._register_class(new_class, key, registry_config)
674 # Handle secondary registrations
675 if registry_config.secondary_registries:
676 mcs._register_secondary(new_class, key, registry_config.secondary_registries)
678 # Log registration if enabled
679 if registry_config.log_registration:
680 logger.debug(f"Auto-registered {name} as '{key}' {registry_config.registry_name}")
682 return new_class
684 @staticmethod
685 def _get_registration_key(
686 name: str,
687 cls: type,
688 config: RegistryConfig,
689 ) -> RegistryKey | None:
690 """Get the registration key for a class (explicit or derived)."""
691 # Try explicit key first
692 key = getattr(cls, config.key_attribute, None)
693 if key is not None:
694 if not isinstance(key, Hashable):
695 raise TypeError(f"Registry key {key!r} declared by {cls.__name__} is not hashable")
696 return key
698 # Try key extractor if provided
699 if config.key_extractor is not None:
700 return config.key_extractor(name, cls)
702 return None
704 @staticmethod
705 def _handle_missing_key(name: str, config: RegistryConfig, new_class: type) -> type:
706 """Handle case where no registration key is available."""
707 if config.skip_if_no_key:
708 if config.log_registration:
709 logger.debug(f"Skipping registration for {name} - no {config.key_attribute}")
710 return new_class # Return the class, just don't register it
711 else:
712 raise ValueError(
713 f"Class {name} must have {config.key_attribute} attribute "
714 f"or provide a key_extractor in registry config"
715 )
717 @classmethod
718 def _auto_configure_registry(
719 cls: type["AutoRegisterMeta"],
720 new_class: type,
721 attrs: dict[str, Any],
722 ) -> RegistryConfig | None:
723 """
724 Auto-configure registry from metaclass OR base class attributes.
726 Priority:
727 1. Metaclass attributes (__registry_dict__, __registry_key__ on metaclass)
728 2. Base class attributes (__registry_key__ on class, auto-create __registry__)
729 3. Parent class __registry__ (inherit from parent)
731 Returns:
732 RegistryConfig if auto-configuration successful, None otherwise
733 """
734 # Check if the metaclass has __registry_dict__ attribute (old style)
735 registry_dict = getattr(cls, "__registry_dict__", None)
737 # If no metaclass registry, check if base class wants auto-creation or inheritance
738 if registry_dict is None:
739 # First check if any parent class has __registry__ (inherit from parent)
740 # This takes priority over creating a new registry
741 for base in new_class.__mro__[1:]: # Skip self
742 if hasattr(base, "__registry__"):
743 registry_dict = getattr(base, "__registry__")
744 key_attribute = getattr(base, "__registry_key__", None)
745 key_extractor = getattr(base, "__key_extractor__", None)
746 skip_if_no_key = getattr(base, "__skip_if_no_key__", True)
747 secondary_registries = getattr(base, "__secondary_registries__", None)
748 registry_name = getattr(base, "__registry_name__", None)
749 break
750 else:
751 # No parent registry found - check if class explicitly defines a
752 # registry family or __registry_key__ (only create new registry
753 # from the class body, or from an inherited registry protocol
754 # mixin that does not own a registry itself).
755 registry_family = attrs.get("__registry_family__")
756 key_attribute = attrs.get("__registry_key__")
757 if key_attribute is None and isinstance(registry_family, RegistryFamily):
758 key_attribute = registry_family.key_attribute_name
759 inherited_registry_base = None
760 if key_attribute is None:
761 for base in new_class.__mro__[1:]:
762 inherited_key_attribute = getattr(base, "__registry_key__", None)
763 if inherited_key_attribute is not None:
764 inherited_registry_base = base
765 key_attribute = inherited_key_attribute
766 break
767 if key_attribute is not None:
768 # Check if class already provides its own __registry__ dict
769 # (allows opting out of LazyDiscoveryDict)
770 if "__registry__" in attrs:
771 registry_dict = attrs["__registry__"]
772 else:
773 # Auto-create registry dict and store on the class
774 registry_dict = LazyDiscoveryDict()
775 setattr(new_class, "__registry__", registry_dict)
777 # Get other optional attributes from class. Explicit legacy
778 # class attributes override the family/inherited declaration.
779 key_extractor = attrs.get("__key_extractor__")
780 if key_extractor is None and inherited_registry_base is not None:
781 key_extractor = getattr(inherited_registry_base, "__key_extractor__", None)
782 if "__skip_if_no_key__" in attrs:
783 skip_if_no_key = attrs["__skip_if_no_key__"]
784 elif isinstance(registry_family, RegistryFamily):
785 skip_if_no_key = registry_family.skip_if_no_key
786 elif inherited_registry_base is not None:
787 skip_if_no_key = getattr(
788 inherited_registry_base, "__skip_if_no_key__", True
789 )
790 else:
791 skip_if_no_key = True
792 secondary_registries = attrs.get("__secondary_registries__")
793 if secondary_registries is None and inherited_registry_base is not None:
794 secondary_registries = getattr(
795 inherited_registry_base, "__secondary_registries__", None
796 )
797 if "__registry_name__" in attrs:
798 registry_name = attrs["__registry_name__"]
799 elif isinstance(registry_family, RegistryFamily):
800 registry_name = registry_family.registry_name
801 elif inherited_registry_base is not None:
802 registry_name = getattr(inherited_registry_base, "__registry_name__", None)
803 else:
804 registry_name = None
805 else:
806 return None # No registry configuration found
807 else:
808 # Old style: get from metaclass
809 key_attribute = getattr(cls, "__registry_key__", "_registry_key")
810 key_extractor = getattr(cls, "__key_extractor__", None)
811 skip_if_no_key = getattr(cls, "__skip_if_no_key__", True)
812 secondary_registries = getattr(cls, "__secondary_registries__", None)
813 registry_name = getattr(cls, "__registry_name__", None)
815 # Auto-derive registry name if not provided
816 if registry_name is None:
817 # Derive from class name: "StorageBackend" → "storage backend"
818 clean_name = new_class.__name__
819 for suffix in ["Base", "Meta", "Handler", "Registry"]:
820 if clean_name.endswith(suffix):
821 clean_name = clean_name[: -len(suffix)]
822 break
823 # Convert CamelCase to space-separated lowercase
824 import re
826 registry_name = re.sub(r"([A-Z])", r" \1", clean_name).strip().lower()
828 logger.debug(
829 f"Auto-configured registry for {new_class.__name__}: "
830 f"key_attribute={key_attribute}, registry_name={registry_name}"
831 )
833 if not isinstance(key_attribute, str):
834 raise TypeError(f"Registry key attribute for {new_class.__name__} must be a string")
836 return RegistryConfig(
837 registry_dict=registry_dict,
838 key_attribute=key_attribute,
839 key_extractor=key_extractor,
840 skip_if_no_key=skip_if_no_key,
841 secondary_registries=secondary_registries,
842 registry_name=registry_name,
843 )
845 @staticmethod
846 def _register_class(cls: type, key: RegistryKey, config: RegistryConfig) -> None:
847 """Register class in primary registry."""
848 config.registry_dict[key] = cls
849 setattr(cls, config.key_attribute, key)
851 @staticmethod
852 def _register_secondary(
853 cls: type,
854 primary_key: RegistryKey,
855 secondary_registries: list[SecondaryRegistry],
856 ) -> None:
857 """Handle secondary registry registrations."""
858 for sec_reg in secondary_registries:
859 value = getattr(cls, sec_reg.attr_name, None)
860 if value is None:
861 continue
863 # Determine the key for secondary registration
864 secondary_key: RegistryKey
865 if sec_reg.key_source == PRIMARY_KEY:
866 secondary_key = primary_key
867 else:
868 declared_secondary_key = getattr(cls, sec_reg.key_source, None)
869 if declared_secondary_key is None:
870 logger.warning(
871 f"Cannot register {sec_reg.attr_name} for {cls.__name__} - "
872 f"no {sec_reg.key_source} attribute"
873 )
874 continue
875 if not isinstance(declared_secondary_key, Hashable):
876 raise TypeError(
877 f"Secondary registry key {declared_secondary_key!r} declared "
878 f"by {cls.__name__} is not hashable"
879 )
880 secondary_key = declared_secondary_key
882 # Register in secondary registry
883 sec_reg.registry_dict[secondary_key] = value
884 logger.debug(
885 f"Auto-registered {sec_reg.attr_name} from {cls.__name__} as '{secondary_key}'"
886 )
889class RegisteredEnumMeta(AutoRegisterMeta, EnumMeta):
890 """Metaclass for enum families that also need AutoRegisterMeta membership."""
893# Helper functions for common key extraction patterns
896def extract_key_from_class_name(name: str, cls: type) -> str:
897 """Use the concrete class name as its registry key."""
898 return name
901def make_suffix_extractor(suffix: str) -> KeyExtractor:
902 """
903 Create a key extractor that removes a suffix from class names.
905 Args:
906 suffix: The suffix to remove (e.g., 'Handler', 'Backend')
908 Returns:
909 A key extractor function
911 Examples:
912 extract_handler = make_suffix_extractor('Handler')
913 extract_handler('ImageXpressHandler', cls) -> 'imagexpress'
915 extract_backend = make_suffix_extractor('Backend')
916 extract_backend('DiskStorageBackend', cls) -> 'diskstorage'
917 """
918 suffix_len = len(suffix)
920 def extractor(name: str, cls: type) -> str:
921 if name.endswith(suffix):
922 return name[:-suffix_len].lower()
923 return name.lower()
925 return extractor
928# Pre-built extractors for common patterns
929extract_key_from_handler_suffix = make_suffix_extractor("Handler")
930extract_key_from_backend_suffix = make_suffix_extractor("Backend")