Handlers

provide.foundation.errors.handlers

TODO: Add module docstring.

Classes

ErrorHandler

Configurable error handler with type-based policies.

Attributes:

Name	Type	Description
policies	dict[type[Exception], Callable[[Exception], Any]]	Mapping of error types to handler functions.
default_action	Callable[[Exception], Any]	Default handler for unmatched errors.
log_all	bool	Whether to log all handled errors.
capture_context	bool	Whether to capture error context.
reraise_unhandled	bool	Whether to re-raise unhandled errors.

Examples:

>>> def handle_validation(e: ValidationError):
...     return {"error": "Invalid input", "details": e.context}
...
>>> handler = ErrorHandler(
...     policies={ValidationError: handle_validation},
...     default_action=lambda e: None
... )
>>> result = handler.handle(some_error)

Functions

add_policy

add_policy(
    error_type: type[Exception],
    handler: Callable[[Exception], Any],
) -> ErrorHandler

Add or update a handler policy for an error type.

Parameters:

Name	Type	Description	Default
error_type	type[Exception]	Exception type to handle.	required
handler	Callable[[Exception], Any]	Handler function for this error type.	required

Returns:

Type	Description
ErrorHandler	Self for method chaining.

Source code in provide/foundation/errors/handlers.py

def add_policy(self, error_type: type[Exception], handler: Callable[[Exception], Any]) -> ErrorHandler:
    """Add or update a handler policy for an error type.

    Args:
        error_type: Exception type to handle.
        handler: Handler function for this error type.

    Returns:
        Self for method chaining.

    """
    self.policies[error_type] = handler
    return self

handle

handle(error: Exception) -> Any

Handle an error based on configured policies.

Parameters:

Name	Type	Description	Default
error	Exception	The exception to handle.	required

Returns:

Type	Description
Any	Result from the handler function.

Raises:

Type	Description
Exception	The original error if reraise_unhandled=True and no handler matches.

Examples:

>>> result = handler.handle(ValidationError("Invalid"))

Source code in provide/foundation/errors/handlers.py

def handle(self, error: Exception) -> Any:
    """Handle an error based on configured policies.

    Args:
        error: The exception to handle.

    Returns:
        Result from the handler function.

    Raises:
        Exception: The original error if reraise_unhandled=True and no handler matches.

    Examples:
        >>> result = handler.handle(ValidationError("Invalid"))

    """
    # Find matching handler
    handler = None
    for error_type, policy_handler in self.policies.items():
        if isinstance(error, error_type):
            handler = policy_handler
            break

    # Use default if no match
    if handler is None:
        handler = self.default_action

        # Check if we should re-raise unhandled
        if self.reraise_unhandled and handler is self.default_action:
            if self.log_all:
                from provide.foundation.hub.foundation import get_foundation_logger

                get_foundation_logger().warning(
                    f"No handler for {type(error).__name__}, re-raising",
                    error=str(error),
                )
            raise error

    # Capture context if configured
    context = None
    if self.capture_context:
        context = capture_error_context(error)

    # Log if configured
    if self.log_all:
        log_context = context.to_dict() if context else {}
        from provide.foundation.hub.foundation import get_foundation_logger

        get_foundation_logger().info(
            f"Handling {type(error).__name__} with {handler.__name__}",
            **log_context,
        )

    # Execute handler
    try:
        return handler(error)
    except Exception as handler_error:
        if self.log_all:
            from provide.foundation.hub.foundation import get_foundation_logger

            get_foundation_logger().error(
                f"Error handler failed: {handler_error}",
                exc_info=True,
                original_error=str(error),
                handler=handler.__name__,
            )
        raise handler_error from error

Functions

create_error_handler

create_error_handler(
    **policies: Callable[[Exception], Any],
) -> ErrorHandler

Create an error handler with policies from keyword arguments.

Parameters:

Name	Type	Description	Default
**policies	Callable[[Exception], Any]	Error type names mapped to handler functions.	{}

Returns:

Type	Description
ErrorHandler	Configured ErrorHandler instance.

Examples:

>>> handler = create_error_handler(
...     ValidationError=lambda e: {"error": str(e)},
...     NetworkError=lambda e: retry_operation(),
...     default=lambda e: None
... )

Source code in provide/foundation/errors/handlers.py

def create_error_handler(**policies: Callable[[Exception], Any]) -> ErrorHandler:
    """Create an error handler with policies from keyword arguments.

    Args:
        **policies: Error type names mapped to handler functions.

    Returns:
        Configured ErrorHandler instance.

    Examples:
        >>> handler = create_error_handler(
        ...     ValidationError=lambda e: {"error": str(e)},
        ...     NetworkError=lambda e: retry_operation(),
        ...     default=lambda e: None
        ... )

    """
    # Extract default if provided
    default = policies.pop("default", lambda e: None)

    # Import error types
    import provide.foundation.errors as errors_module

    # Build policies dict
    error_policies = {}
    for error_name, handler_func in policies.items():
        # Try to get the error class from errors module
        error_class = getattr(errors_module, error_name, None)
        if error_class and isinstance(error_class, type) and issubclass(error_class, Exception):
            error_policies[error_class] = handler_func
        else:
            from provide.foundation.hub.foundation import get_foundation_logger

            get_foundation_logger().warning(f"Unknown error type: {error_name}")

    return ErrorHandler(policies=error_policies, default_action=default)

error_boundary

error_boundary(
    *catch: type[Exception],
    on_error: Callable[[Exception], Any] | None = None,
    log_errors: bool = True,
    reraise: bool = True,
    context: dict[str, Any] | None = None,
    fallback: Any = None
) -> Generator[None, None, None]

Context manager for structured error handling with logging.

Parameters:

Name	Type	Description	Default
*catch	type[Exception]	Exception types to catch (defaults to Exception if empty).	()
on_error	Callable[[Exception], Any] | None	Optional callback function when error is caught.	None
log_errors	bool	Whether to log caught errors.	True
reraise	bool	Whether to re-raise after handling.	True
context	dict[str, Any] | None	Additional context for error logging.	None
fallback	Any	Value to return if error is suppressed (when reraise=False).	None

Yields:

Type	Description
None	None

Examples:

>>> with error_boundary(ValueError, on_error=handle_error):
...     risky_operation()

>>> # Suppress and log specific errors
>>> with error_boundary(KeyError, reraise=False, fallback=None):
...     value = data["missing_key"]

Source code in provide/foundation/errors/handlers.py

@contextmanager
def error_boundary(
    *catch: type[Exception],
    on_error: Callable[[Exception], Any] | None = None,
    log_errors: bool = True,
    reraise: bool = True,
    context: dict[str, Any] | None = None,
    fallback: Any = None,
) -> Generator[None, None, None]:
    """Context manager for structured error handling with logging.

    Args:
        *catch: Exception types to catch (defaults to Exception if empty).
        on_error: Optional callback function when error is caught.
        log_errors: Whether to log caught errors.
        reraise: Whether to re-raise after handling.
        context: Additional context for error logging.
        fallback: Value to return if error is suppressed (when reraise=False).

    Yields:
        None

    Examples:
        >>> with error_boundary(ValueError, on_error=handle_error):
        ...     risky_operation()

        >>> # Suppress and log specific errors
        >>> with error_boundary(KeyError, reraise=False, fallback=None):
        ...     value = data["missing_key"]

    """
    # Default to catching all exceptions if none specified
    catch_types = catch if catch else (Exception,)

    try:
        yield
    except catch_types as e:
        if log_errors:
            # Build error context
            error_context = context or {}

            # Add error details
            error_context.update(
                {
                    "error.type": type(e).__name__,
                    "error.message": str(e),
                },
            )

            # If it's a FoundationError, merge its context
            if isinstance(e, FoundationError) and e.context:
                error_context.update(e.context)

            # Log the error
            from provide.foundation.hub.foundation import get_foundation_logger

            get_foundation_logger().error(f"Error caught in boundary: {e}", exc_info=True, **error_context)

        # Call error handler if provided
        if on_error:
            try:
                on_error(e)
            except Exception as handler_error:
                if log_errors:
                    from provide.foundation.hub.foundation import get_foundation_logger

                    get_foundation_logger().error(
                        f"Error handler failed: {handler_error}",
                        exc_info=True,
                        original_error=str(e),
                    )

        # Re-raise if configured
        if reraise:
            raise

        # Return fallback value if not re-raising
        return fallback

handle_error

handle_error(
    error: Exception,
    *,
    log: bool = True,
    capture_context: bool = True,
    reraise: bool = False,
    fallback: Any = None
) -> Any

Handle an error with logging and optional context capture.

Parameters:

Name	Type	Description	Default
error	Exception	The exception to handle.	required
log	bool	Whether to log the error.	True
capture_context	bool	Whether to capture error context.	True
reraise	bool	Whether to re-raise the error after handling.	False
fallback	Any	Value to return if not re-raising.	None

Returns:

Type	Description
Any	The fallback value if not re-raising.

Raises:

Type	Description
Exception	The original error if reraise=True.

Examples:

>>> try:
...     risky_operation()
... except ValueError as e:
...     result = handle_error(e, fallback="default")

Source code in provide/foundation/errors/handlers.py

def handle_error(
    error: Exception,
    *,
    log: bool = True,
    capture_context: bool = True,
    reraise: bool = False,
    fallback: Any = None,
) -> Any:
    """Handle an error with logging and optional context capture.

    Args:
        error: The exception to handle.
        log: Whether to log the error.
        capture_context: Whether to capture error context.
        reraise: Whether to re-raise the error after handling.
        fallback: Value to return if not re-raising.

    Returns:
        The fallback value if not re-raising.

    Raises:
        Exception: The original error if reraise=True.

    Examples:
        >>> try:
        ...     risky_operation()
        ... except ValueError as e:
        ...     result = handle_error(e, fallback="default")

    """
    # Capture context if requested
    context = None
    if capture_context:
        context = capture_error_context(error)

    # Log if requested
    if log:
        log_context = context.to_dict() if context else {}
        from provide.foundation.hub.foundation import get_foundation_logger

        get_foundation_logger().error(f"Handling error: {error}", exc_info=True, **log_context)

    # Re-raise if requested
    if reraise:
        raise error

    return fallback

transactional

transactional(
    rollback: Callable[[], None],
    commit: Callable[[], None] | None = None,
    on_error: Callable[[Exception], None] | None = None,
    log_errors: bool = True,
) -> Generator[None, None, None]

Context manager for transactional operations with rollback.

Parameters:

Name	Type	Description	Default
rollback	Callable[[], None]	Function to call on error to rollback changes.	required
commit	Callable[[], None] | None	Optional function to call on success.	None
on_error	Callable[[Exception], None] | None	Optional error handler before rollback.	None
log_errors	bool	Whether to log errors.	True

Yields:

Type	Description
None	None

Examples:

>>> def rollback_changes():
...     db.rollback()
...
>>> def commit_changes():
...     db.commit()
...
>>> with transactional(rollback_changes, commit_changes):
...     db.execute("INSERT INTO users ...")
...     db.execute("UPDATE accounts ...")

Source code in provide/foundation/errors/handlers.py

@contextmanager
def transactional(
    rollback: Callable[[], None],
    commit: Callable[[], None] | None = None,
    on_error: Callable[[Exception], None] | None = None,
    log_errors: bool = True,
) -> Generator[None, None, None]:
    """Context manager for transactional operations with rollback.

    Args:
        rollback: Function to call on error to rollback changes.
        commit: Optional function to call on success.
        on_error: Optional error handler before rollback.
        log_errors: Whether to log errors.

    Yields:
        None

    Examples:
        >>> def rollback_changes():
        ...     db.rollback()
        ...
        >>> def commit_changes():
        ...     db.commit()
        ...
        >>> with transactional(rollback_changes, commit_changes):
        ...     db.execute("INSERT INTO users ...")
        ...     db.execute("UPDATE accounts ...")

    """
    try:
        yield
        # Call commit if provided and no exception occurred
        if commit:
            commit()
    except Exception as e:
        if log_errors:
            from provide.foundation.hub.foundation import get_foundation_logger

            get_foundation_logger().error("Transaction failed, rolling back", exc_info=True, error=str(e))

        # Call error handler if provided
        if on_error:
            try:
                on_error(e)
            except Exception as handler_error:
                if log_errors:
                    from provide.foundation.hub.foundation import get_foundation_logger

                    get_foundation_logger().error(
                        f"Transaction error handler failed: {handler_error}",
                        original_error=str(e),
                    )

        # Perform rollback
        try:
            rollback()
            if log_errors:
                from provide.foundation.hub.foundation import get_foundation_logger
            get_foundation_logger().info("Transaction rolled back successfully")
        except Exception as rollback_error:
            if log_errors:
                from provide.foundation.hub.foundation import get_foundation_logger

                get_foundation_logger().critical(f"Rollback failed: {rollback_error}", original_error=str(e))
            # Re-raise the rollback error as it's more critical
            raise rollback_error from e

        # Re-raise original exception
        raise