Skip to content

Event handlers

provide.foundation.hub.event_handlers

TODO: Add module docstring.

Classes

Functions

handle_circuit_breaker_event 

handle_circuit_breaker_event(event: Event) -> None

Handle circuit breaker events by logging them.

Parameters:

Name Type Description Default
event Event

Circuit breaker event to handle

 required
Source code in provide/foundation/hub/event_handlers.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def handle_circuit_breaker_event(event: Event) -> None:
    """Handle circuit breaker events by logging them.

    Args:
        event: Circuit breaker event to handle
    """
    logger = _get_logger_safely()
    if not logger:
        return

    if event.name == "circuit_breaker.recovered":
        logger.info("Circuit breaker recovered - closing circuit", data=event.data)
    elif event.name == "circuit_breaker.opened":
        logger.error("Circuit breaker opened due to failures", data=event.data)
    elif event.name == "circuit_breaker.recovery_failed":
        logger.warning("Circuit breaker recovery failed - opening circuit", data=event.data)
    elif event.name == "circuit_breaker.attempting_recovery":
        logger.info("Circuit breaker attempting recovery", data=event.data)
    elif event.name == "circuit_breaker.manual_reset":
        logger.info("Circuit breaker manually reset", data=event.data)

handle_registry_event 

handle_registry_event(event: Event | RegistryEvent) -> None

Handle registry events by logging them.

Parameters:

Name Type Description Default
event Event | RegistryEvent

Registry event to handle

 required
Source code in provide/foundation/hub/event_handlers.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def handle_registry_event(event: Event | RegistryEvent) -> None:
    """Handle registry events by logging them.

    Args:
        event: Registry event to handle
    """
    logger = _get_logger_safely()
    if not logger:
        return

    if isinstance(event, RegistryEvent):
        if event.operation == "register":
            logger.debug(
                "Registered item",
                item_name=event.item_name,
                dimension=event.dimension,
                data=event.data,
            )
        elif event.operation == "remove":
            logger.debug(
                "Removed item",
                item_name=event.item_name,
                dimension=event.dimension,
                data=event.data,
            )
    elif event.name.startswith("registry."):
        logger.debug("Registry event", event_name=event.name, data=event.data)

set_reset_in_progress 

set_reset_in_progress(in_progress: bool) -> None

Set whether a reset is currently in progress.

This prevents event handlers from triggering logger operations during resets, which would cause infinite loops.

Parameters:

Name Type Description Default
in_progress bool

True if reset is starting, False if reset is complete

 required
Source code in provide/foundation/hub/event_handlers.py 
50
51
52
53
54
55
56
57
58
59
60
def set_reset_in_progress(in_progress: bool) -> None:
    """Set whether a reset is currently in progress.

    This prevents event handlers from triggering logger operations during resets,
    which would cause infinite loops.

    Args:
        in_progress: True if reset is starting, False if reset is complete
    """
    global _reset_in_progress
    _reset_in_progress = in_progress

setup_event_logging 

setup_event_logging() -> None

Set up event handlers to connect events back to logging.

This should be called after the logger is initialized to avoid circular dependencies.

Source code in provide/foundation/hub/event_handlers.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def setup_event_logging() -> None:
    """Set up event handlers to connect events back to logging.

    This should be called after the logger is initialized to avoid
    circular dependencies.
    """
    event_bus = get_event_bus()

    # Subscribe to registry events
    event_bus.subscribe("registry.register", handle_registry_event)
    event_bus.subscribe("registry.remove", handle_registry_event)

    # Subscribe to circuit breaker events
    event_bus.subscribe("circuit_breaker.recovered", handle_circuit_breaker_event)
    event_bus.subscribe("circuit_breaker.opened", handle_circuit_breaker_event)
    event_bus.subscribe("circuit_breaker.recovery_failed", handle_circuit_breaker_event)
    event_bus.subscribe("circuit_breaker.attempting_recovery", handle_circuit_breaker_event)
    event_bus.subscribe("circuit_breaker.manual_reset", handle_circuit_breaker_event)