Skip to content

Handlers

🤖 AI-Generated Content

This documentation was generated with AI assistance and is still being audited. Some, or potentially a lot, of this information may be inaccurate. Learn more.

provide.foundation.hub.handlers

Classes

Functions

execute_error_handlers 

execute_error_handlers(
    exception: Exception, context: dict[str, Any]
) -> dict[str, Any] | None

Execute error handlers until one handles the exception.

Source code in provide/foundation/hub/handlers.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@resilient(
    fallback=None,
    context_provider=lambda: {
        "function": "execute_error_handlers",
        "module": "hub.handlers",
    },
)
def execute_error_handlers(exception: Exception, context: dict[str, Any]) -> dict[str, Any] | None:
    """Execute error handlers until one handles the exception."""
    handlers = get_handlers_for_exception(exception)

    for entry in handlers:
        handler = entry.value
        try:
            result = handler(exception, context)
            if result is not None:
                return result
        except Exception as handler_error:
            get_foundation_logger().error("Error handler failed", handler=entry.name, error=str(handler_error))

    return None

get_handlers_for_exception 

get_handlers_for_exception(
    exception: Exception,
) -> list[RegistryEntry]

Get error handlers that can handle the given exception type.

Source code in provide/foundation/hub/handlers.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def get_handlers_for_exception(exception: Exception) -> list[RegistryEntry]:
    """Get error handlers that can handle the given exception type."""
    registry, ComponentCategory = _get_registry_and_lock()

    # Get all error handlers
    all_entries = list(registry)
    handlers = [entry for entry in all_entries if entry.dimension == ComponentCategory.ERROR_HANDLER.value]

    # Filter by exception type
    exception_type_name = type(exception).__name__
    matching_handlers = []

    for entry in handlers:
        exception_types = entry.metadata.get("exception_types", [])
        if any(
            exc_type in exception_type_name or exception_type_name in exc_type for exc_type in exception_types
        ):
            matching_handlers.append(entry)

    # Sort by priority (highest first)
    matching_handlers.sort(key=lambda e: e.metadata.get("priority", 0), reverse=True)
    return matching_handlers