Coverage for src / metaclass_registry / core.py: 80%
303 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-10 22:33 +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. AutoRegisterMeta applies the configuration during class creation
393. Domain-specific metaclasses provide thin wrappers with their config
41This maintains domain-specific features while eliminating duplication.
42"""
44import importlib
45import logging
46import threading
47from abc import ABCMeta
48from dataclasses import dataclass
49from typing import Dict, Type, Optional, Callable, Any
51logger = logging.getLogger(__name__)
53# Lazy import to avoid circular dependency
54_registry_cache_manager = None
56def _get_cache_manager():
57 """Lazy import of RegistryCacheManager to avoid circular imports."""
58 global _registry_cache_manager
59 if _registry_cache_manager is None:
60 from metaclass_registry.cache import (
61 RegistryCacheManager,
62 CacheConfig,
63 serialize_plugin_class,
64 deserialize_plugin_class,
65 get_package_file_mtimes
66 )
67 _registry_cache_manager = {
68 'RegistryCacheManager': RegistryCacheManager,
69 'CacheConfig': CacheConfig,
70 'serialize_plugin_class': serialize_plugin_class,
71 'deserialize_plugin_class': deserialize_plugin_class,
72 'get_package_file_mtimes': get_package_file_mtimes
73 }
74 return _registry_cache_manager
77# Type aliases for clarity
78RegistryDict = Dict[str, Type]
79KeyExtractor = Callable[[str, Type], str]
81# Constants for key sources
82PRIMARY_KEY = 'primary'
85class SecondaryRegistryDict(dict):
86 """
87 Dict for secondary registries that auto-triggers primary registry discovery.
89 When accessed, this dict triggers discovery of the primary registry,
90 which populates both the primary and secondary registries.
91 """
93 def __init__(self, primary_registry: 'LazyDiscoveryDict'):
94 super().__init__()
95 self._primary_registry = primary_registry
97 def _ensure_discovered(self):
98 """Trigger discovery of primary registry (which populates this secondary registry)."""
99 if hasattr(self._primary_registry, '_discover'):
100 self._primary_registry._discover()
102 def __getitem__(self, key):
103 self._ensure_discovered()
104 return super().__getitem__(key)
106 def __contains__(self, key):
107 self._ensure_discovered()
108 return super().__contains__(key)
110 def __iter__(self):
111 self._ensure_discovered()
112 return super().__iter__()
114 def __len__(self):
115 self._ensure_discovered()
116 return super().__len__()
118 def keys(self):
119 self._ensure_discovered()
120 return super().keys()
122 def values(self):
123 self._ensure_discovered()
124 return super().values()
126 def items(self):
127 self._ensure_discovered()
128 return super().items()
130 def get(self, key, default=None):
131 self._ensure_discovered()
132 return super().get(key, default)
135class LazyDiscoveryDict(dict):
136 """
137 Dict that auto-discovers plugins on first access with optional caching.
139 Supports caching discovered plugins to speed up subsequent application starts.
140 Cache is validated against package version and file modification times.
142 Thread-safe: Uses locking to ensure discovery happens only once
143 even when accessed from multiple threads simultaneously.
144 """
146 def __init__(self, enable_cache: bool = True):
147 """
148 Initialize lazy discovery dict.
150 Args:
151 enable_cache: If True, use caching to speed up discovery
152 """
153 super().__init__()
154 self._base_class = None
155 self._config = None
156 self._discovered = False
157 self._enable_cache = enable_cache
158 self._cache_manager = None
159 self._discovery_lock = threading.RLock() # Reentrant lock for same-thread re-entry
161 def _set_config(self, base_class: Type, config: 'RegistryConfig') -> None:
162 self._base_class = base_class
163 self._config = config
165 # Initialize cache manager if caching is enabled
166 if self._enable_cache and config.discovery_package:
167 try:
168 cache_utils = _get_cache_manager()
170 # Get version getter (try to get version from discovery package)
171 def get_version():
172 try:
173 # Try to get version from the root package
174 root_package = config.discovery_package.split('.')[0]
175 mod = __import__(root_package)
176 return getattr(mod, '__version__', 'unknown')
177 except:
178 return "unknown"
180 self._cache_manager = cache_utils['RegistryCacheManager'](
181 cache_name=f"{config.registry_name.replace(' ', '_')}_registry",
182 version_getter=get_version,
183 serializer=cache_utils['serialize_plugin_class'],
184 deserializer=cache_utils['deserialize_plugin_class'],
185 config=cache_utils['CacheConfig'](
186 max_age_days=7,
187 check_mtimes=True # Validate file modifications
188 )
189 )
190 except Exception as e:
191 logger.debug(f"Failed to initialize cache manager: {e}")
192 self._cache_manager = None
194 def _discover(self) -> None:
195 """
196 Run discovery once, using cache if available.
198 Thread-safe: Discovery happens inside the lock to ensure atomicity.
199 The lock is held for the entire duration to prevent other threads
200 from reading a partially-populated registry.
202 CRITICAL: No fast path check outside the lock! All threads must acquire
203 the lock to ensure they don't read a partially-populated registry.
204 """
205 # No config = nothing to discover
206 if not self._config or not self._config.discovery_package:
207 return
209 # ALWAYS acquire lock - no fast path to avoid race condition
210 # RLock allows same thread to re-acquire during module imports
211 with self._discovery_lock:
212 # Check if already discovered (inside lock)
213 if self._discovered:
214 return
216 # Mark as discovered to prevent infinite re-entry from same thread
217 # (module imports during discovery might access registry)
218 self._discovered = True
220 # Try to load from cache first
221 if self._cache_manager:
222 try:
223 cached_plugins = self._cache_manager.load_cache()
224 if cached_plugins is not None:
225 # Reconstruct registry from cache
226 self.update(cached_plugins)
227 logger.info(
228 f"✅ Loaded {len(self)} {self._config.registry_name}s from cache"
229 )
230 return
231 except Exception as e:
232 logger.debug(f"Cache load failed for {self._config.registry_name}: {e}")
234 # Cache miss or disabled - perform full discovery
235 try:
236 pkg = importlib.import_module(self._config.discovery_package)
238 if self._config.discovery_function:
239 self._config.discovery_function(
240 pkg.__path__,
241 f"{self._config.discovery_package}.",
242 self._base_class
243 )
244 else:
245 from metaclass_registry.discovery import (
246 discover_registry_classes,
247 discover_registry_classes_recursive
248 )
249 func = (
250 discover_registry_classes_recursive
251 if self._config.discovery_recursive
252 else discover_registry_classes
253 )
254 func(pkg.__path__, f"{self._config.discovery_package}.", self._base_class)
256 logger.debug(f"Discovered {len(self)} {self._config.registry_name}s")
258 # Save to cache if enabled
259 if self._cache_manager:
260 try:
261 cache_utils = _get_cache_manager()
262 file_mtimes = cache_utils['get_package_file_mtimes'](
263 self._config.discovery_package
264 )
265 self._cache_manager.save_cache(dict(self), file_mtimes)
266 except Exception as e:
267 logger.debug(f"Failed to save cache for {self._config.registry_name}: {e}")
269 except Exception as e:
270 logger.warning(f"Discovery failed: {e}")
271 # Lock released here - registry is now fully populated and safe to read
273 def __getitem__(self, k):
274 with self._discovery_lock:
275 self._discover()
276 return super().__getitem__(k)
278 def __contains__(self, k):
279 with self._discovery_lock:
280 self._discover()
281 return super().__contains__(k)
283 def __iter__(self):
284 with self._discovery_lock:
285 self._discover()
286 return super().__iter__()
288 def __len__(self):
289 with self._discovery_lock:
290 self._discover()
291 return super().__len__()
293 def keys(self):
294 with self._discovery_lock:
295 self._discover()
296 return super().keys()
298 def values(self):
299 with self._discovery_lock:
300 self._discover()
301 return super().values()
303 def items(self):
304 with self._discovery_lock:
305 self._discover()
306 return super().items()
308 def get(self, k, default=None):
309 with self._discovery_lock:
310 self._discover()
311 return super().get(k, default)
314@dataclass(frozen=True)
315class SecondaryRegistry:
316 """Configuration for a secondary registry (e.g., metadata handlers)."""
317 registry_dict: RegistryDict
318 key_source: str # 'primary' or attribute name
319 attr_name: str # Attribute to check on the class
322@dataclass(frozen=True)
323class RegistryConfig:
324 """
325 Configuration for automatic class registration behavior.
327 This dataclass encapsulates all the configuration needed for metaclass
328 registration, making the pattern explicit and easy to understand.
330 Attributes:
331 registry_dict: Dictionary to register classes into (e.g., MICROSCOPE_HANDLERS)
332 key_attribute: Name of class attribute containing the registration key
333 (e.g., '_microscope_type', '_backend_type', '_context_type')
334 key_extractor: Optional function to derive key from class name if key_attribute
335 is not set. Signature: (class_name: str, cls: Type) -> str
336 skip_if_no_key: If True, skip registration when key_attribute is None.
337 If False, require either key_attribute or key_extractor.
338 secondary_registries: Optional list of secondary registry configurations
339 log_registration: If True, log debug message when class is registered
340 registry_name: Human-readable name for logging (e.g., 'microscope handler')
341 discovery_package: Optional package name to auto-discover (e.g., 'openhcs.microscopes')
342 discovery_recursive: If True, use recursive discovery (default: False)
344 Examples:
345 # Microscope handlers with name-based key extraction and secondary registry
346 RegistryConfig(
347 registry_dict=MICROSCOPE_HANDLERS,
348 key_attribute='_microscope_type',
349 key_extractor=extract_key_from_handler_suffix,
350 skip_if_no_key=False,
351 secondary_registries=[
352 SecondaryRegistry(
353 registry_dict=METADATA_HANDLERS,
354 key_source=PRIMARY_KEY,
355 attr_name='_metadata_handler_class'
356 )
357 ],
358 log_registration=True,
359 registry_name='microscope handler'
360 )
362 # Storage backends with explicit key and skip-if-none behavior
363 RegistryConfig(
364 registry_dict=STORAGE_BACKENDS,
365 key_attribute='_backend_type',
366 skip_if_no_key=True,
367 registry_name='storage backend'
368 )
370 # Context providers with simple explicit key
371 RegistryConfig(
372 registry_dict=CONTEXT_PROVIDERS,
373 key_attribute='_context_type',
374 skip_if_no_key=True,
375 registry_name='context provider'
376 )
377 """
378 registry_dict: RegistryDict
379 key_attribute: str
380 key_extractor: Optional[KeyExtractor] = None
381 skip_if_no_key: bool = False
382 secondary_registries: Optional[list[SecondaryRegistry]] = None
383 log_registration: bool = True
384 registry_name: str = "plugin"
385 discovery_package: Optional[str] = None # Auto-inferred from base class module if None
386 discovery_recursive: bool = False
387 discovery_function: Optional[Callable] = None # Custom discovery function
390class AutoRegisterMeta(ABCMeta):
391 """
392 Generic metaclass for automatic plugin registration (Pattern A).
394 This metaclass automatically registers concrete classes in a global registry
395 when they are defined, eliminating the need for manual registration calls.
397 Features:
398 - Skips abstract classes (checks __abstractmethods__)
399 - Supports explicit keys via class attributes
400 - Supports derived keys via key extraction functions
401 - Supports secondary registries (e.g., metadata handlers)
402 - Configurable skip-if-no-key behavior
403 - Debug logging for registration events
405 Usage:
406 # Create domain-specific metaclass
407 class MicroscopeHandlerMeta(AutoRegisterMeta):
408 def __new__(mcs, name, bases, attrs):
409 return super().__new__(mcs, name, bases, attrs,
410 registry_config=_MICROSCOPE_REGISTRY_CONFIG)
412 # Use in class definition
413 class ImageXpressHandler(MicroscopeHandler, metaclass=MicroscopeHandlerMeta):
414 _microscope_type = 'imagexpress' # Optional if key_extractor is provided
415 _metadata_handler_class = ImageXpressMetadata # Optional secondary registration
417 Design Principles:
418 - Explicit configuration over magic behavior
419 - Preserve all domain-specific features
420 - Zero breaking changes to existing code
421 - Easy to understand and debug
422 """
424 def __new__(mcs, name: str, bases: tuple, attrs: dict,
425 registry_config: Optional[RegistryConfig] = None):
426 """
427 Create a new class and register it if appropriate.
429 Args:
430 name: Name of the class being created
431 bases: Base classes
432 attrs: Class attributes dictionary
433 registry_config: Configuration for registration behavior.
434 If None, auto-configures from class attributes or skips registration.
436 Returns:
437 The newly created class
438 """
439 # Create the class using ABCMeta
440 new_class = super().__new__(mcs, name, bases, attrs)
442 # Auto-configure registry if not provided but class has __registry__ attributes
443 if registry_config is None:
444 registry_config = mcs._auto_configure_registry(new_class, attrs)
445 if registry_config is None:
446 return new_class # No config and no auto-config possible
448 # Set up lazy discovery if registry dict supports it (only once for base class)
449 if isinstance(registry_config.registry_dict, LazyDiscoveryDict) and not registry_config.registry_dict._config:
450 from dataclasses import replace
452 # Auto-infer discovery_package from base class module if not specified
453 config = registry_config
454 if config.discovery_package is None:
455 # Extract package from base class module (e.g., 'openhcs.microscopes.microscope_base' → 'openhcs.microscopes')
456 module_parts = new_class.__module__.rsplit('.', 1)
457 inferred_package = module_parts[0] if len(module_parts) > 1 else new_class.__module__
458 # Create new config with inferred package
459 config = replace(config, discovery_package=inferred_package)
460 logger.debug(f"Auto-inferred discovery_package='{inferred_package}' from {new_class.__module__}")
462 # Auto-infer discovery_recursive based on package structure
463 # Check if package has subdirectories with __init__.py (indicating nested structure)
464 if config.discovery_package:
465 try:
466 pkg = importlib.import_module(config.discovery_package)
467 if hasattr(pkg, '__path__'):
468 import os
469 has_subpackages = False
470 for path in pkg.__path__:
471 if os.path.isdir(path):
472 # Check if any subdirectories contain __init__.py
473 for entry in os.listdir(path):
474 subdir = os.path.join(path, entry)
475 if os.path.isdir(subdir) and os.path.exists(os.path.join(subdir, '__init__.py')):
476 has_subpackages = True
477 break
478 if has_subpackages:
479 break
481 # Only override if discovery_recursive is still at default (False)
482 # This allows explicit overrides to take precedence
483 if has_subpackages and not config.discovery_recursive:
484 config = replace(config, discovery_recursive=True)
485 logger.debug(f"Auto-inferred discovery_recursive=True for '{config.discovery_package}' (has subpackages)")
486 elif not has_subpackages and config.discovery_recursive:
487 logger.debug(f"Keeping explicit discovery_recursive=True for '{config.discovery_package}' (no subpackages detected)")
488 except Exception as e:
489 logger.debug(f"Failed to auto-infer discovery_recursive: {e}")
491 # Auto-wrap secondary registries with SecondaryRegistryDict
492 if config.secondary_registries:
493 import sys
494 wrapped_secondaries = []
495 module = sys.modules.get(new_class.__module__)
497 for sec_reg in config.secondary_registries:
498 # Check if secondary registry needs wrapping
499 if isinstance(sec_reg.registry_dict, dict) and not isinstance(sec_reg.registry_dict, SecondaryRegistryDict):
500 # Create a new SecondaryRegistryDict wrapping the primary registry
501 wrapped_dict = SecondaryRegistryDict(registry_config.registry_dict)
502 # Copy any existing entries from the old dict
503 wrapped_dict.update(sec_reg.registry_dict)
505 # Find and update the module global variable
506 if module:
507 for var_name, var_value in vars(module).items():
508 if var_value is sec_reg.registry_dict:
509 setattr(module, var_name, wrapped_dict)
510 logger.debug(f"Auto-wrapped secondary registry '{var_name}' in {new_class.__module__}")
511 break
513 # Create new SecondaryRegistry with wrapped dict
514 wrapped_sec_reg = SecondaryRegistry(
515 registry_dict=wrapped_dict,
516 key_source=sec_reg.key_source,
517 attr_name=sec_reg.attr_name
518 )
519 wrapped_secondaries.append(wrapped_sec_reg)
520 else:
521 wrapped_secondaries.append(sec_reg)
523 # Rebuild config with wrapped secondary registries
524 config = replace(config, secondary_registries=wrapped_secondaries)
526 registry_config.registry_dict._set_config(new_class, config)
528 # Only register concrete classes (not abstract base classes)
529 if not bases or getattr(new_class, '__abstractmethods__', None):
530 return new_class
532 # Get or derive the registration key
533 key = mcs._get_registration_key(name, new_class, registry_config)
535 # Handle missing key
536 if key is None:
537 return mcs._handle_missing_key(name, registry_config, new_class)
539 # Register in primary registry
540 mcs._register_class(new_class, key, registry_config)
542 # Handle secondary registrations
543 if registry_config.secondary_registries:
544 mcs._register_secondary(new_class, key, registry_config.secondary_registries)
546 # Log registration if enabled
547 if registry_config.log_registration:
548 logger.debug(f"Auto-registered {name} as '{key}' {registry_config.registry_name}")
550 return new_class
552 @staticmethod
553 def _get_registration_key(name: str, cls: Type, config: RegistryConfig) -> Optional[str]:
554 """Get the registration key for a class (explicit or derived)."""
555 # Try explicit key first
556 key = getattr(cls, config.key_attribute, None)
557 if key is not None:
558 return key
560 # Try key extractor if provided
561 if config.key_extractor is not None:
562 return config.key_extractor(name, cls)
564 return None
566 @staticmethod
567 def _handle_missing_key(name: str, config: RegistryConfig, new_class: Type) -> Type:
568 """Handle case where no registration key is available."""
569 if config.skip_if_no_key:
570 if config.log_registration:
571 logger.debug(f"Skipping registration for {name} - no {config.key_attribute}")
572 return new_class # Return the class, just don't register it
573 else:
574 raise ValueError(
575 f"Class {name} must have {config.key_attribute} attribute "
576 f"or provide a key_extractor in registry config"
577 )
579 @classmethod
580 def _auto_configure_registry(mcs, new_class: Type, attrs: dict) -> Optional[RegistryConfig]:
581 """
582 Auto-configure registry from metaclass OR base class attributes.
584 Priority:
585 1. Metaclass attributes (__registry_dict__, __registry_key__ on metaclass)
586 2. Base class attributes (__registry_key__ on class, auto-create __registry__)
587 3. Parent class __registry__ (inherit from parent)
589 Returns:
590 RegistryConfig if auto-configuration successful, None otherwise
591 """
592 # Check if the metaclass has __registry_dict__ attribute (old style)
593 registry_dict = getattr(mcs, '__registry_dict__', None)
595 # If no metaclass registry, check if base class wants auto-creation or inheritance
596 if registry_dict is None:
597 # First check if any parent class has __registry__ (inherit from parent)
598 # This takes priority over creating a new registry
599 for base in new_class.__mro__[1:]: # Skip self
600 if hasattr(base, '__registry__'):
601 registry_dict = base.__registry__
602 key_attribute = getattr(base, '__registry_key__', None)
603 key_extractor = getattr(base, '__key_extractor__', None)
604 skip_if_no_key = getattr(base, '__skip_if_no_key__', True)
605 secondary_registries = getattr(base, '__secondary_registries__', None)
606 registry_name = getattr(base, '__registry_name__', None)
607 break
608 else:
609 # No parent registry found - check if class explicitly defines __registry_key__
610 # (only create new registry if __registry_key__ is in the class body, not inherited)
611 key_attribute = attrs.get('__registry_key__')
612 if key_attribute is not None:
613 # Check if class already provides its own __registry__ dict
614 # (allows opting out of LazyDiscoveryDict)
615 if '__registry__' in attrs:
616 registry_dict = attrs['__registry__']
617 else:
618 # Auto-create registry dict and store on the class
619 registry_dict = LazyDiscoveryDict()
620 new_class.__registry__ = registry_dict
622 # Get other optional attributes from class
623 key_extractor = attrs.get('__key_extractor__')
624 skip_if_no_key = attrs.get('__skip_if_no_key__', True)
625 secondary_registries = attrs.get('__secondary_registries__')
626 registry_name = attrs.get('__registry_name__')
627 else:
628 return None # No registry configuration found
629 else:
630 # Old style: get from metaclass
631 key_attribute = getattr(mcs, '__registry_key__', '_registry_key')
632 key_extractor = getattr(mcs, '__key_extractor__', None)
633 skip_if_no_key = getattr(mcs, '__skip_if_no_key__', True)
634 secondary_registries = getattr(mcs, '__secondary_registries__', None)
635 registry_name = getattr(mcs, '__registry_name__', None)
637 # Auto-derive registry name if not provided
638 if registry_name is None:
639 # Derive from class name: "StorageBackend" → "storage backend"
640 clean_name = new_class.__name__
641 for suffix in ['Base', 'Meta', 'Handler', 'Registry']:
642 if clean_name.endswith(suffix):
643 clean_name = clean_name[:-len(suffix)]
644 break
645 # Convert CamelCase to space-separated lowercase
646 import re
647 registry_name = re.sub(r'([A-Z])', r' \1', clean_name).strip().lower()
649 logger.debug(f"Auto-configured registry for {new_class.__name__}: "
650 f"key_attribute={key_attribute}, registry_name={registry_name}")
652 return RegistryConfig(
653 registry_dict=registry_dict,
654 key_attribute=key_attribute,
655 key_extractor=key_extractor,
656 skip_if_no_key=skip_if_no_key,
657 secondary_registries=secondary_registries,
658 registry_name=registry_name
659 )
661 @staticmethod
662 def _register_class(cls: Type, key: str, config: RegistryConfig) -> None:
663 """Register class in primary registry."""
664 config.registry_dict[key] = cls
665 setattr(cls, config.key_attribute, key)
667 @staticmethod
668 def _register_secondary(
669 cls: Type,
670 primary_key: str,
671 secondary_registries: list[SecondaryRegistry]
672 ) -> None:
673 """Handle secondary registry registrations."""
674 for sec_reg in secondary_registries:
675 value = getattr(cls, sec_reg.attr_name, None)
676 if value is None:
677 continue
679 # Determine the key for secondary registration
680 if sec_reg.key_source == PRIMARY_KEY:
681 secondary_key = primary_key
682 else:
683 secondary_key = getattr(cls, sec_reg.key_source, None)
684 if secondary_key is None:
685 logger.warning(
686 f"Cannot register {sec_reg.attr_name} for {cls.__name__} - "
687 f"no {sec_reg.key_source} attribute"
688 )
689 continue
691 # Register in secondary registry
692 sec_reg.registry_dict[secondary_key] = value
693 logger.debug(f"Auto-registered {sec_reg.attr_name} from {cls.__name__} as '{secondary_key}'")
696# Helper functions for common key extraction patterns
698def make_suffix_extractor(suffix: str) -> KeyExtractor:
699 """
700 Create a key extractor that removes a suffix from class names.
702 Args:
703 suffix: The suffix to remove (e.g., 'Handler', 'Backend')
705 Returns:
706 A key extractor function
708 Examples:
709 extract_handler = make_suffix_extractor('Handler')
710 extract_handler('ImageXpressHandler', cls) -> 'imagexpress'
712 extract_backend = make_suffix_extractor('Backend')
713 extract_backend('DiskStorageBackend', cls) -> 'diskstorage'
714 """
715 suffix_len = len(suffix)
717 def extractor(name: str, cls: Type) -> str:
718 if name.endswith(suffix):
719 return name[:-suffix_len].lower()
720 return name.lower()
722 return extractor
725# Pre-built extractors for common patterns
726extract_key_from_handler_suffix = make_suffix_extractor('Handler')
727extract_key_from_backend_suffix = make_suffix_extractor('Backend')