API Reference

Core Module

Generic metaclass infrastructure for automatic plugin registration.

This module provides reusable metaclass infrastructure for Pattern A registry systems (1:1 class-to-plugin mapping with automatic discovery). It eliminates code duplication across MicroscopeHandlerMeta, StorageBackendMeta, and ContextProviderMeta.

Pattern Selection Guide:

Use AutoRegisterMeta (Pattern A) when: - You have a 1:1 mapping between classes and plugins - Plugins should be automatically discovered and registered - Registration happens at class definition time - Simple metadata (just a key and maybe one secondary registry)

Use Service Pattern (Pattern B) when: - You have many-to-one mapping (multiple items per plugin) - Complex metadata (FunctionMetadata with 8+ fields) - Need aggregation across multiple sources - Examples: Function registry, Format registry

Use Functional Registry (Pattern C) when: - Simple type-to-handler mappings - No state needed - Functional programming style preferred - Examples: Widget creation registries

Use Manual Registration (Pattern D) when: - Complex initialization logic required - Explicit control over registration timing needed - Very few plugins (< 3) - Examples: ZMQ servers, Pipeline steps

Architecture:

AutoRegisterMeta uses a configuration-driven approach: 1. RegistryConfig defines registration behavior 2. AutoRegisterMeta applies the configuration during class creation 3. Domain-specific metaclasses provide thin wrappers with their config

This maintains domain-specific features while eliminating duplication.

class metaclass_registry.core.SecondaryRegistryDict(primary_registry: LazyDiscoveryDict)

Bases: dict

Dict for secondary registries that auto-triggers primary registry discovery.

When accessed, this dict triggers discovery of the primary registry, which populates both the primary and secondary registries.

__init__(primary_registry: LazyDiscoveryDict)

keys() → a set-like object providing a view on D's keys

values() → an object providing a view on D's values

items() → a set-like object providing a view on D's items

get(key, default=None)

Return the value for key if key is in the dictionary, else default.

class metaclass_registry.core.LazyDiscoveryDict(enable_cache: bool = True)

Bases: dict

Dict that auto-discovers plugins on first access with optional caching.

Supports caching discovered plugins to speed up subsequent application starts. Cache is validated against package version and file modification times.

Thread-safe: Uses locking to ensure discovery happens only once even when accessed from multiple threads simultaneously.

__init__(enable_cache: bool = True)

Initialize lazy discovery dict.

Parameters:

enable_cache – If True, use caching to speed up discovery

keys() → a set-like object providing a view on D's keys

values() → an object providing a view on D's values

items() → a set-like object providing a view on D's items

get(k, default=None)

Return the value for key if key is in the dictionary, else default.

__getstate__()

Return state for pickling, excluding non-picklable objects.

The _discovery_lock and _cache_manager are excluded since they contain non-picklable objects (RLock and potentially closures).

__setstate__(state)

Restore state from pickle, recreating non-picklable objects.

The _discovery_lock is recreated as a new RLock. The _cache_manager is set to None and will be reinitialized if needed.

class metaclass_registry.core.SecondaryRegistry(registry_dict: Dict[str, Type], key_source: str, attr_name: str)

Bases: object

Configuration for a secondary registry (e.g., metadata handlers).

registry_dict: Dict[str, Type]

key_source: str

attr_name: str

__init__(registry_dict: Dict[str, Type], key_source: str, attr_name: str) → None

class metaclass_registry.core.RegistryConfig(registry_dict: Dict[str, Type], key_attribute: str, key_extractor: Callable[[str, Type], str] | None = None, skip_if_no_key: bool = False, secondary_registries: list[SecondaryRegistry] | None = None, log_registration: bool = True, registry_name: str = 'plugin', discovery_package: str | None = None, discovery_recursive: bool = False, discovery_function: Callable | None = None)

Bases: object

Configuration for automatic class registration behavior.

This dataclass encapsulates all the configuration needed for metaclass registration, making the pattern explicit and easy to understand.

registry_dict

Dictionary to register classes into (e.g., MICROSCOPE_HANDLERS)

Type:

Dict[str, Type]

key_attribute

Name of class attribute containing the registration key (e.g., ‘_microscope_type’, ‘_backend_type’, ‘_context_type’)

Type:

str

key_extractor

Optional function to derive key from class name if key_attribute is not set. Signature: (class_name: str, cls: Type) -> str

Type:

Callable[[str, Type], str] | None

skip_if_no_key

If True, skip registration when key_attribute is None. If False, require either key_attribute or key_extractor.

Type:

bool

secondary_registries

Optional list of secondary registry configurations

Type:

list[metaclass_registry.core.SecondaryRegistry] | None

log_registration

If True, log debug message when class is registered

Type:

bool

registry_name

Human-readable name for logging (e.g., ‘microscope handler’)

Type:

str

discovery_package

Optional package name to auto-discover (e.g., ‘openhcs.microscopes’)

Type:

str | None

discovery_recursive

If True, use recursive discovery (default: False)

Type:

bool

Examples

# Microscope handlers with name-based key extraction and secondary registry RegistryConfig(

registry_dict=MICROSCOPE_HANDLERS, key_attribute=’_microscope_type’, key_extractor=extract_key_from_handler_suffix, skip_if_no_key=False, secondary_registries=[

SecondaryRegistry(

registry_dict=METADATA_HANDLERS, key_source=PRIMARY_KEY, attr_name=’_metadata_handler_class’

)

], log_registration=True, registry_name=’microscope handler’

)

# Storage backends with explicit key and skip-if-none behavior RegistryConfig(

registry_dict=STORAGE_BACKENDS, key_attribute=’_backend_type’, skip_if_no_key=True, registry_name=’storage backend’

)

# Context providers with simple explicit key RegistryConfig(

registry_dict=CONTEXT_PROVIDERS, key_attribute=’_context_type’, skip_if_no_key=True, registry_name=’context provider’

)

registry_dict: Dict[str, Type]

key_attribute: str

key_extractor: Callable[[str, Type], str] | None = None

skip_if_no_key: bool = False

secondary_registries: list[SecondaryRegistry] | None = None

log_registration: bool = True

registry_name: str = 'plugin'

discovery_package: str | None = None

discovery_recursive: bool = False

discovery_function: Callable | None = None

__init__(registry_dict: Dict[str, Type], key_attribute: str, key_extractor: Callable[[str, Type], str] | None = None, skip_if_no_key: bool = False, secondary_registries: list[SecondaryRegistry] | None = None, log_registration: bool = True, registry_name: str = 'plugin', discovery_package: str | None = None, discovery_recursive: bool = False, discovery_function: Callable | None = None) → None

class metaclass_registry.core.AutoRegisterMeta(name: str, bases: tuple, attrs: dict, registry_config: RegistryConfig | None = None)

Bases: ABCMeta

Generic metaclass for automatic plugin registration (Pattern A).

This metaclass automatically registers concrete classes in a global registry when they are defined, eliminating the need for manual registration calls.

Features: - Skips abstract classes (checks __abstractmethods__) - Supports explicit keys via class attributes - Supports derived keys via key extraction functions - Supports secondary registries (e.g., metadata handlers) - Configurable skip-if-no-key behavior - Debug logging for registration events

Usage:

# Create domain-specific metaclass class MicroscopeHandlerMeta(AutoRegisterMeta):

def __new__(mcs, name, bases, attrs):

return super().__new__(mcs, name, bases, attrs,

registry_config=_MICROSCOPE_REGISTRY_CONFIG)

# Use in class definition class ImageXpressHandler(MicroscopeHandler, metaclass=MicroscopeHandlerMeta):

_microscope_type = ‘imagexpress’ # Optional if key_extractor is provided _metadata_handler_class = ImageXpressMetadata # Optional secondary registration

Design Principles: - Explicit configuration over magic behavior - Preserve all domain-specific features - Zero breaking changes to existing code - Easy to understand and debug

static __new__(mcs, name: str, bases: tuple, attrs: dict, registry_config: RegistryConfig | None = None)

Create a new class and register it if appropriate.

Parameters:

• name – Name of the class being created

• bases – Base classes

• attrs – Class attributes dictionary

• registry_config – Configuration for registration behavior. If None, auto-configures from class attributes or skips registration.

Returns:

The newly created class

metaclass_registry.core.make_suffix_extractor(suffix: str) → Callable[[str, Type], str]

Create a key extractor that removes a suffix from class names.

Parameters:

suffix – The suffix to remove (e.g., ‘Handler’, ‘Backend’)

Returns:

A key extractor function

Examples

extract_handler = make_suffix_extractor(‘Handler’) extract_handler(‘ImageXpressHandler’, cls) -> ‘imagexpress’

extract_backend = make_suffix_extractor(‘Backend’) extract_backend(‘DiskStorageBackend’, cls) -> ‘diskstorage’

metaclass_registry.core.extract_key_from_handler_suffix(name: str, cls: Type) → str

metaclass_registry.core.extract_key_from_backend_suffix(name: str, cls: Type) → str

Discovery Module

Generic registry class discovery utility.

Consolidates duplicated registry discovery patterns across: - Library registries (processing backends) - Format registries (experimental analysis) - Microscope handler registries - Storage backend registries

This module eliminates ~70 lines of duplicated pkgutil + importlib boilerplate by providing a single, well-tested discovery function.

metaclass_registry.discovery.discover_registry_classes(package_path: Iterable[str], package_prefix: str, base_class: Type, exclude_modules: Set[str] | None = None, validation_func: Callable[[Type], bool] | None = None, skip_packages: bool = True) → List[Type]

Generic registry class discovery using pkgutil + importlib pattern.

Scans a package for classes that inherit from a base class and automatically discovers them for registration. This eliminates duplicated discovery code across different registry systems.

Parameters:

• package_path – Package __path__ attribute to scan (e.g., openhcs.io.__path__) Accepts any iterable of strings (List, Tuple, _NamespacePath, etc.)

• package_prefix – Module prefix for importlib (e.g., “openhcs.io.”)

• base_class – Base class to filter for (e.g., StorageBackend)

• exclude_modules – Set of module name substrings to skip (e.g., {‘base’, ‘registry’})

• validation_func – Optional function to validate discovered classes Should return True to include, False to exclude

• skip_packages – If True, skip package directories (default: True)

Returns:

List of discovered registry classes

Example

>>> from openhcs.io.base import StorageBackend
>>> import openhcs.io
>>> backends = discover_registry_classes(
...     package_path=openhcs.io.__path__,
...     package_prefix="openhcs.io.",
...     base_class=StorageBackend,
...     exclude_modules={'base', 'backend_registry'}
... )
>>> print([b.__name__ for b in backends])
['DiskStorageBackend', 'MemoryStorageBackend', 'ZarrStorageBackend']

metaclass_registry.discovery.discover_registry_classes_recursive(package_path: Iterable[str], package_prefix: str, base_class: Type, exclude_modules: Set[str] | None = None, validation_func: Callable[[Type], bool] | None = None) → List[Type]

Recursive version of discover_registry_classes that walks entire package tree.

Uses pkgutil.walk_packages instead of iter_modules to recursively scan all subpackages. Useful for deeply nested registry structures.

Parameters:

• package_path – Package __path__ attribute to scan Accepts any iterable of strings (List, Tuple, _NamespacePath, etc.)

• package_prefix – Module prefix for importlib

• base_class – Base class to filter for

• exclude_modules – Set of module name substrings to skip

• validation_func – Optional function to validate discovered classes

Returns:

List of discovered registry classes

Example

>>> from openhcs.processing.backends.lib_registry.unified_registry import LibraryRegistryBase
>>> import openhcs.processing.backends.experimental_analysis
>>> registries = discover_registry_classes_recursive(
...     package_path=openhcs.processing.backends.experimental_analysis.__path__,
...     package_prefix="openhcs.processing.backends.experimental_analysis.",
...     base_class=MicroscopeFormatRegistryBase,
...     exclude_modules={'base'}
... )

Cache Module

Generic caching system for plugin registries.

Provides unified caching for both function registries (Pattern B) and metaclass registries (Pattern A), eliminating code duplication and ensuring consistent cache behavior across the codebase.

Architecture: - RegistryCacheManager: Generic cache manager for any registry type - Supports version validation, age-based invalidation, mtime checking - JSON-based serialization with custom serializers/deserializers - XDG-compliant cache locations

Usage:

# For function registries cache_mgr = RegistryCacheManager(

cache_name=”scikit_image_functions”, version_getter=lambda: skimage.__version__, serializer=serialize_function_metadata, deserializer=deserialize_function_metadata

)

# For metaclass registries cache_mgr = RegistryCacheManager(

cache_name=”microscope_handlers”, version_getter=lambda: openhcs.__version__, serializer=serialize_plugin_class, deserializer=deserialize_plugin_class

)

metaclass_registry.cache.get_cache_file_path(cache_name: str) → Path

Get XDG-compliant cache file path.

Parameters:

cache_name – Name of the cache file

Returns:

Path to cache file in XDG cache directory

class metaclass_registry.cache.CacheConfig(max_age_days: int = 7, check_mtimes: bool = False, cache_version: str = '1.0')

Bases: object

Configuration for registry caching behavior.

max_age_days: int = 7

check_mtimes: bool = False

cache_version: str = '1.0'

__init__(max_age_days: int = 7, check_mtimes: bool = False, cache_version: str = '1.0') → None

class metaclass_registry.cache.RegistryCacheManager(cache_name: str, version_getter: Callable[[], str], serializer: Callable[[T], Dict[str, Any]], deserializer: Callable[[Dict[str, Any]], T], config: CacheConfig | None = None)

Bases: Generic[T]

Generic cache manager for plugin registries.

Handles caching, validation, and reconstruction of registry data with support for version checking, age-based invalidation, and custom serialization.

Type Parameters:

T: Type of items being cached (e.g., FunctionMetadata, Type[Plugin])

__init__(cache_name: str, version_getter: Callable[[], str], serializer: Callable[[T], Dict[str, Any]], deserializer: Callable[[Dict[str, Any]], T], config: CacheConfig | None = None)

Initialize cache manager.

Parameters:

• cache_name – Name for the cache file (e.g., “microscope_handlers”)

• version_getter – Function that returns current version string

• serializer – Function to serialize item to JSON-compatible dict

• deserializer – Function to deserialize dict back to item

• config – Optional cache configuration

load_cache() → Dict[str, T] | None

Load cached items with validation.

Returns:

Dictionary of cached items, or None if cache is invalid

save_cache(items: Dict[str, T], file_mtimes: Dict[str, float] | None = None) → None

Save items to cache.

Parameters:

• items – Dictionary of items to cache

• file_mtimes – Optional dict of file paths to modification times

clear_cache() → None

Clear the cache file.

metaclass_registry.cache.serialize_plugin_class(plugin_class: type) → Dict[str, Any]

Serialize a plugin class to JSON-compatible dict.

Parameters:

plugin_class – Plugin class to serialize

Returns:

Dictionary with module and class name

metaclass_registry.cache.deserialize_plugin_class(data: Dict[str, Any]) → type

Deserialize a plugin class from JSON-compatible dict.

Parameters:

data – Dictionary with module and class name

Returns:

Reconstructed plugin class

Raises:

• ImportError – If module cannot be imported

• AttributeError – If class not found in module

metaclass_registry.cache.get_package_file_mtimes(package_path: str) → Dict[str, float]

Get modification times for all Python files in a package.

Parameters:

package_path – Package path (e.g., “openhcs.microscopes”)

Returns:

Dictionary mapping file paths to modification times

Exceptions Module

Exceptions for metaclass-registry.

exception metaclass_registry.exceptions.RegistryError

Bases: Exception

Base exception for registry-related errors.

exception metaclass_registry.exceptions.DiscoveryError

Bases: RegistryError

Exception raised when plugin discovery fails.

exception metaclass_registry.exceptions.CacheError

Bases: RegistryError

Exception raised when cache operations fail.