Registry

provide.foundation.eventsets.registry

TODO: Add module docstring.

Classes

EventSetRegistry

EventSetRegistry()

Bases: Registry

Registry for event set definitions using foundation Registry.

Extends the foundation Registry to provide specialized methods for event set registration and discovery.

Source code in provide/foundation/hub/registry.py

def __init__(self) -> None:
    """Initialize an empty registry."""
    # Use managed lock for deadlock prevention
    # Lock is registered during Foundation initialization via register_foundation_locks()
    from provide.foundation.concurrency.locks import get_lock_manager

    self._lock = get_lock_manager().get_lock("foundation.registry")
    self._registry: dict[str, dict[str, RegistryEntry]] = defaultdict(dict)
    self._aliases: dict[str, tuple[str, str]] = {}
    # Type-based registry for dependency injection
    self._type_registry: dict[type[Any], Any] = {}

Functions

discover_sets

discover_sets() -> None

Auto-discover and register event sets from the sets/ directory.

Imports all modules in the sets/ subdirectory and registers any EVENT_SET constants found.

Source code in provide/foundation/eventsets/registry.py

def discover_sets(self) -> None:
    """Auto-discover and register event sets from the sets/ directory.

    Imports all modules in the sets/ subdirectory and registers
    any EVENT_SET constants found.
    """
    sets_path = Path(__file__).parent / "sets"
    if not sets_path.exists():
        logger.debug("No sets directory found for auto-discovery")
        return

    for module_info in pkgutil.iter_modules([str(sets_path)]):
        if module_info.ispkg:
            continue

        module_name = f"provide.foundation.eventsets.sets.{module_info.name}"
        try:
            module = importlib.import_module(module_name)

            if hasattr(module, "EVENT_SET"):
                event_set = module.EVENT_SET
                if isinstance(event_set, EventSet):
                    try:
                        self.register_event_set(event_set)
                        logger.debug(
                            "Auto-discovered event set",
                            module=module_name,
                            name=event_set.name,
                        )
                    except AlreadyExistsError:
                        logger.trace(
                            "Event set already registered during discovery",
                            module=module_name,
                            name=event_set.name,
                        )
                else:
                    logger.warning(
                        "EVENT_SET is not an EventSet",
                        module=module_name,
                        type=type(event_set).__name__,
                    )

        except ImportError as e:
            logger.debug(
                "Failed to import event set module",
                module=module_name,
                error=str(e),
            )
        except Exception as e:
            logger.warning(
                "Error during event set discovery",
                module=module_name,
                error=str(e),
                error_type=type(e).__name__,
            )

get_event_set

get_event_set(name: str) -> EventSet

Retrieve an event set by name.

Parameters:

Name	Type	Description	Default
name	str	The name of the event set	required

Returns:

Type	Description
EventSet	The EventSet

Raises:

Type	Description
NotFoundError	If no event set with this name exists

Source code in provide/foundation/eventsets/registry.py

def get_event_set(self, name: str) -> EventSet:
    """Retrieve an event set by name.

    Args:
        name: The name of the event set

    Returns:
        The EventSet

    Raises:
        NotFoundError: If no event set with this name exists

    """
    event_set = self.get(name, "eventset")
    if event_set is None:
        raise NotFoundError(f"Event set '{name}' not found")
    return event_set

list_event_sets

list_event_sets() -> list[EventSet]

List all registered event sets sorted by priority.

Returns:

Type	Description
list[EventSet]	List of EventSet objects sorted by descending priority

Source code in provide/foundation/eventsets/registry.py

def list_event_sets(self) -> list[EventSet]:
    """List all registered event sets sorted by priority.

    Returns:
        List of EventSet objects sorted by descending priority

    """
    names = self.list_dimension("eventset")
    entries = [self.get_entry(name, "eventset") for name in names]
    valid_entries = [entry for entry in entries if entry is not None]
    valid_entries.sort(key=lambda e: e.metadata.get("priority", 0), reverse=True)
    return [entry.value for entry in valid_entries]

register_event_set

register_event_set(event_set: EventSet) -> None

Register an event set definition.

Parameters:

Name	Type	Description	Default
event_set	EventSet	The EventSet to register	required

Raises:

Type	Description
AlreadyExistsError	If an event set with this name already exists

Source code in provide/foundation/eventsets/registry.py

def register_event_set(self, event_set: EventSet) -> None:
    """Register an event set definition.

    Args:
        event_set: The EventSet to register

    Raises:
        AlreadyExistsError: If an event set with this name already exists

    """
    try:
        self.register(
            event_set.name,
            event_set,
            "eventset",
            metadata={"priority": event_set.priority},
        )
        logger.debug(
            "Registered event set",
            name=event_set.name,
            priority=event_set.priority,
            field_count=len(event_set.field_mappings),
            mapping_count=len(event_set.mappings),
        )
    except AlreadyExistsError:
        logger.trace("Event set already registered", name=event_set.name)
        raise

Functions

clear_registry

clear_registry() -> None

Clear the registry for testing.

Source code in provide/foundation/eventsets/registry.py

def clear_registry() -> None:
    """Clear the registry for testing."""
    global _registry, _discovery_completed
    _registry = EventSetRegistry()
    _discovery_completed = False
    logger.trace("Event set registry cleared")

discover_event_sets

discover_event_sets() -> None

Auto-discover and register all event sets.

Source code in provide/foundation/eventsets/registry.py

def discover_event_sets() -> None:
    """Auto-discover and register all event sets."""
    global _discovery_completed
    if _discovery_completed:
        logger.trace("Event set discovery already completed, skipping")
        return

    logger.debug("Starting event set discovery")
    _registry.discover_sets()
    _discovery_completed = True
    logger.debug("Event set discovery completed")

get_registry

get_registry() -> EventSetRegistry

Get the global event set registry instance.

Source code in provide/foundation/eventsets/registry.py

def get_registry() -> EventSetRegistry:
    """Get the global event set registry instance."""
    return _registry

register_event_set

register_event_set(event_set: EventSet) -> None

Register an event set in the global registry.

Parameters:

Name	Type	Description	Default
event_set	EventSet	The EventSet to register	required

Source code in provide/foundation/eventsets/registry.py

def register_event_set(event_set: EventSet) -> None:
    """Register an event set in the global registry.

    Args:
        event_set: The EventSet to register

    """
    _registry.register_event_set(event_set)

reset_discovery_state

reset_discovery_state() -> None

Reset discovery state for testing.

Source code in provide/foundation/eventsets/registry.py

def reset_discovery_state() -> None:
    """Reset discovery state for testing."""
    global _discovery_completed
    _discovery_completed = False
    logger.trace("Event set discovery state reset")