Errors

provide.foundation.errors

TODO: Add module docstring.

Classes

AlreadyExistsError

AlreadyExistsError(
    message: str,
    *,
    resource_type: str | None = None,
    resource_id: str | None = None,
    **kwargs: Any
)

Bases: FoundationError

Raised when attempting to create a resource that already exists.

Parameters:

Name	Type	Description	Default
message	str	Error message describing the conflict.	required
resource_type	str | None	Optional type of resource.	None
resource_id	str | None	Optional resource identifier.	None
**kwargs	Any	Additional context passed to FoundationError.	{}

Examples:

>>> raise AlreadyExistsError("User already registered")
>>> raise AlreadyExistsError("Duplicate key", resource_type="user", resource_id="[email protected]")

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

def __init__(
    self,
    message: str,
    *,
    resource_type: str | None = None,
    resource_id: str | None = None,
    **kwargs: Any,
) -> None:
    if resource_type:
        kwargs.setdefault("context", {})["exists.type"] = resource_type
    if resource_id:
        kwargs.setdefault("context", {})["exists.id"] = resource_id
    super().__init__(message, **kwargs)

AuthenticationError

AuthenticationError(
    message: str,
    *,
    auth_method: str | None = None,
    realm: str | None = None,
    **kwargs: Any
)

Bases: FoundationError

Raised when authentication fails.

Parameters:

Name	Type	Description	Default
message	str	Error message describing the authentication failure.	required
auth_method	str | None	Optional authentication method used.	None
realm	str | None	Optional authentication realm.	None
**kwargs	Any	Additional context passed to FoundationError.	{}

Examples:

>>> raise AuthenticationError("Invalid credentials")
>>> raise AuthenticationError("Token expired", auth_method="jwt")

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

def __init__(
    self,
    message: str,
    *,
    auth_method: str | None = None,
    realm: str | None = None,
    **kwargs: Any,
) -> None:
    if auth_method:
        kwargs.setdefault("context", {})["auth.method"] = auth_method
    if realm:
        kwargs.setdefault("context", {})["auth.realm"] = realm
    super().__init__(message, **kwargs)

AuthorizationError

AuthorizationError(
    message: str,
    *,
    required_permission: str | None = None,
    resource: str | None = None,
    actor: str | None = None,
    **kwargs: Any
)

Bases: FoundationError

Raised when authorization fails.

Parameters:

Name	Type	Description	Default
message	str	Error message describing the authorization failure.	required
required_permission	str | None	Optional required permission.	None
resource	str | None	Optional resource being accessed.	None
actor	str | None	Optional actor (user/service) attempting access.	None
**kwargs	Any	Additional context passed to FoundationError.	{}

Examples:

>>> raise AuthorizationError("Access denied")
>>> raise AuthorizationError("Insufficient permissions", required_permission="admin")

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

def __init__(
    self,
    message: str,
    *,
    required_permission: str | None = None,
    resource: str | None = None,
    actor: str | None = None,
    **kwargs: Any,
) -> None:
    if required_permission:
        kwargs.setdefault("context", {})["authz.permission"] = required_permission
    if resource:
        kwargs.setdefault("context", {})["authz.resource"] = resource
    if actor:
        kwargs.setdefault("context", {})["authz.actor"] = actor
    super().__init__(message, **kwargs)

CommandNotFoundError

CommandNotFoundError(
    message: str,
    *,
    command: str | list[str] | None = None,
    return_code: int | None = None,
    stdout: str | bytes | None = None,
    stderr: str | bytes | None = None,
    timeout: bool = False,
    code: str | None = None,
    **extra_context: Any
)

Bases: ProcessError

Error when a command/executable is not found.

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

def __init__(
    self,
    message: str,
    *,
    command: str | list[str] | None = None,
    return_code: int | None = None,
    stdout: str | bytes | None = None,
    stderr: str | bytes | None = None,
    timeout: bool = False,
    code: str | None = None,
    **extra_context: Any,
) -> None:
    """Initialize ProcessError with command execution details.

    Args:
        message: Human-readable error message
        command: The command that was executed
        return_code: Process return/exit code
        stdout: Standard output from the process
        stderr: Standard error from the process
        timeout: Whether the process timed out
        code: Optional error code
        **extra_context: Additional context information

    """
    # Build comprehensive error message
    full_message = message

    if command:
        cmd_str = command if isinstance(command, str) else " ".join(command)
        full_message += f"\nCommand: {cmd_str}"

    if return_code is not None:
        full_message += f"\nReturn code: {return_code}"

    if timeout:
        full_message += "\nProcess timed out"

    if stdout:
        stdout_str = stdout.decode("utf-8", "replace") if isinstance(stdout, bytes) else stdout
        if stdout_str.strip():
            full_message += f"\n--- STDOUT ---\n{stdout_str.strip()}"

    if stderr:
        stderr_str = stderr.decode("utf-8", "replace") if isinstance(stderr, bytes) else stderr
        if stderr_str.strip():
            full_message += f"\n--- STDERR ---\n{stderr_str.strip()}"

    # Store structured data
    context = extra_context.copy()
    context.update(
        {
            "process.command": command,
            "process.return_code": return_code,
            "process.timeout": timeout,
        },
    )

    # Store clean stdout/stderr for programmatic access
    self.stdout = (
        stdout.decode("utf-8", "replace").strip()
        if isinstance(stdout, bytes)
        else stdout.strip()
        if stdout
        else None
    )

    self.stderr = (
        stderr.decode("utf-8", "replace").strip()
        if isinstance(stderr, bytes)
        else stderr.strip()
        if stderr
        else None
    )

    self.command = command
    self.return_code = return_code
    self.timeout = timeout

    super().__init__(full_message, code=code, context=context)

ConcurrencyError

ConcurrencyError(
    message: str,
    *,
    conflict_type: str | None = None,
    version_expected: Any = None,
    version_actual: Any = None,
    **kwargs: Any
)

Bases: FoundationError

Raised when concurrency conflicts occur.

Parameters:

Name	Type	Description	Default
message	str	Error message describing the concurrency issue.	required
conflict_type	str | None	Optional type of conflict (lock, version, etc.).	None
version_expected	Any	Optional expected version.	None
version_actual	Any	Optional actual version.	None
**kwargs	Any	Additional context passed to FoundationError.	{}

Examples:

>>> raise ConcurrencyError("Optimistic lock failure")
>>> raise ConcurrencyError("Version mismatch", version_expected=1, version_actual=2)

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

def __init__(
    self,
    message: str,
    *,
    conflict_type: str | None = None,
    version_expected: Any = None,
    version_actual: Any = None,
    **kwargs: Any,
) -> None:
    if conflict_type:
        kwargs.setdefault("context", {})["concurrency.type"] = conflict_type
    if version_expected is not None:
        kwargs.setdefault("context", {})["concurrency.version_expected"] = str(version_expected)
    if version_actual is not None:
        kwargs.setdefault("context", {})["concurrency.version_actual"] = str(version_actual)
    super().__init__(message, **kwargs)

ConfigValidationError

ConfigValidationError(
    message: str,
    *,
    config_class: str | None = None,
    **kwargs: Any
)

Bases: ValidationError

Raised when configuration validation fails.

This is a specialized validation error for configuration-specific validation failures.

Parameters:

Name	Type	Description	Default
message	str	Validation error message.	required
config_class	str | None	Optional name of the config class.	None
**kwargs	Any	Additional context passed to ValidationError.	{}

Examples:

>>> raise ConfigValidationError("Invalid database configuration")
>>> raise ConfigValidationError("Port must be positive", field="port", value=-1)

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

def __init__(
    self,
    message: str,
    *,
    config_class: str | None = None,
    **kwargs: Any,
) -> None:
    if config_class:
        kwargs.setdefault("context", {})["config.class"] = config_class
    super().__init__(message, **kwargs)

ConfigurationError

ConfigurationError(
    message: str,
    *,
    config_key: str | None = None,
    config_source: str | None = None,
    **kwargs: Any
)

Bases: FoundationError

Raised when configuration is invalid or cannot be loaded.

Parameters:

Name	Type	Description	Default
message	str	Error message describing the configuration issue.	required
config_key	str | None	Optional configuration key that caused the error.	None
config_source	str | None	Optional source of the configuration (file, env, etc.).	None
**kwargs	Any	Additional context passed to FoundationError.	{}

Examples:

>>> raise ConfigurationError("Missing required config")
>>> raise ConfigurationError("Invalid timeout", config_key="timeout")

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

def __init__(
    self,
    message: str,
    *,
    config_key: str | None = None,
    config_source: str | None = None,
    **kwargs: Any,
) -> None:
    if config_key:
        kwargs.setdefault("context", {})["config.key"] = config_key
    if config_source:
        kwargs.setdefault("context", {})["config.source"] = config_source
    super().__init__(message, **kwargs)

DependencyError

DependencyError(
    package: str,
    *,
    feature: str | None = None,
    install_command: str | None = None,
    **kwargs: Any
)

Bases: FoundationError

Raised when an optional dependency is required but not installed.

Parameters:

Name	Type	Description	Default
package	str	Name of the missing package	required
feature	str | None	Optional feature name that requires the package	None
install_command	str | None	Optional custom installation command	None
**kwargs	Any	Additional context passed to FoundationError	{}

Examples:

>>> raise DependencyError("cryptography", feature="crypto")
>>> raise DependencyError("requests", install_command="pip install requests")

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

def __init__(
    self,
    package: str,
    *,
    feature: str | None = None,
    install_command: str | None = None,
    **kwargs: Any,
) -> None:
    # Determine the installation command
    # Feature takes priority over custom install_command
    if feature:
        cmd = f"pip install 'provide-foundation[{feature}]'"
    elif install_command:
        cmd = install_command
    else:
        cmd = f"pip install {package}"

    # Create the error message
    message = f"Optional dependency '{package}' is required for this feature. Install with: {cmd}"

    # Add context
    context = kwargs.setdefault("context", {})
    context["dependency.package"] = package
    context["dependency.install_command"] = cmd
    if feature:
        context["dependency.feature"] = feature

    super().__init__(message, **kwargs)

DependencyMismatchError

DependencyMismatchError(
    package: str,
    *,
    required_version: str,
    current_version: str,
    **kwargs: Any
)

Bases: FoundationError

Raised when a dependency version doesn't meet requirements.

Parameters:

Name	Type	Description	Default
package	str	Name of the package with version mismatch	required
required_version	str	Required version or constraint	required
current_version	str	Currently installed version	required
**kwargs	Any	Additional context passed to FoundationError	{}

Examples:

>>> raise DependencyMismatchError("cryptography", ">=3.0.0", "2.9.2")

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

def __init__(
    self,
    package: str,
    *,
    required_version: str,
    current_version: str,
    **kwargs: Any,
) -> None:
    message = (
        f"Package '{package}' version {current_version} does not meet "
        f"requirement {required_version}. Please upgrade with: "
        f"pip install '{package}{required_version}'"
    )

    # Add context
    context = kwargs.setdefault("context", {})
    context["dependency.package"] = package
    context["dependency.required_version"] = required_version
    context["dependency.current_version"] = current_version

    super().__init__(message, **kwargs)

ErrorCategory

Bases: str, Enum

Error categorization for routing and handling.

ErrorCode

Bases: str, Enum

Standard error codes for programmatic error handling.

Use these codes for consistent error identification across the system. Codes are grouped by category with prefixes: - CFG: Configuration errors - VAL: Validation errors - INT: Integration errors - RES: Resource errors - AUTH: Authentication/Authorization errors - SYS: System errors

ErrorContext

Rich error context for diagnostics and monitoring.

Provides a flexible container for error metadata that can be used by different systems (logging, monitoring, Terraform, etc.).

Attributes:

Name	Type	Description
timestamp	datetime	When the error occurred.
severity	ErrorSeverity	Error severity level.
category	ErrorCategory	Error category for classification.
metadata	dict[str, dict[str, Any]]	Namespace-based metadata storage.
tags	set[str]	Set of tags for categorization and filtering.
trace_id	str | None	Optional trace ID for distributed tracing.
span_id	str | None	Optional span ID for distributed tracing.

Examples:

>>> ctx = ErrorContext(severity=ErrorSeverity.HIGH)
>>> ctx.add_namespace("aws", {"region": "us-east-1", "account": "123456"})
>>> ctx.add_tag("production")

Functions

add_namespace

add_namespace(
    namespace: str, data: dict[str, Any]
) -> ErrorContext

Add namespaced metadata.

Parameters:

Name	Type	Description	Default
namespace	str	Namespace key (e.g., 'terraform', 'aws', 'http').	required
data	dict[str, Any]	Metadata for this namespace.	required

Returns:

Type	Description
ErrorContext	Self for method chaining.

Examples:

>>> ctx.add_namespace("terraform", {"provider": "aws", "version": "5.0"})
>>> ctx.add_namespace("http", {"method": POST, "status": 500})

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

def add_namespace(self, namespace: str, data: dict[str, Any]) -> ErrorContext:
    """Add namespaced metadata.

    Args:
        namespace: Namespace key (e.g., 'terraform', 'aws', 'http').
        data: Metadata for this namespace.

    Returns:
        Self for method chaining.

    Examples:
        >>> ctx.add_namespace("terraform", {"provider": "aws", "version": "5.0"})
        >>> ctx.add_namespace("http", {"method": POST, "status": 500})

    """
    self.metadata[namespace] = data
    return self

add_tag

add_tag(tag: str) -> ErrorContext

Add a tag for categorization.

Parameters:

Name	Type	Description	Default
tag	str	Tag to add.	required

Returns:

Type	Description
ErrorContext	Self for method chaining.

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

def add_tag(self, tag: str) -> ErrorContext:
    """Add a tag for categorization.

    Args:
        tag: Tag to add.

    Returns:
        Self for method chaining.

    """
    self.tags.add(tag)
    return self

add_tags

add_tags(*tags: str) -> ErrorContext

Add multiple tags.

Parameters:

Name	Type	Description	Default
*tags	str	Tags to add.	()

Returns:

Type	Description
ErrorContext	Self for method chaining.

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

def add_tags(self, *tags: str) -> ErrorContext:
    """Add multiple tags.

    Args:
        *tags: Tags to add.

    Returns:
        Self for method chaining.

    """
    self.tags.update(tags)
    return self

get_namespace

get_namespace(namespace: str) -> dict[str, Any] | None

Get metadata for a specific namespace.

Parameters:

Name	Type	Description	Default
namespace	str	Namespace key to retrieve.	required

Returns:

Type	Description
dict[str, Any] | None	Namespace metadata or None if not found.

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

def get_namespace(self, namespace: str) -> dict[str, Any] | None:
    """Get metadata for a specific namespace.

    Args:
        namespace: Namespace key to retrieve.

    Returns:
        Namespace metadata or None if not found.

    """
    return self.metadata.get(namespace)

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary for logging and serialization.

Returns:

Type	Description
dict[str, Any]	Flattened dictionary with namespaced keys.

Examples:

>>> ctx.to_dict()
{'timestamp': '2024-01-01T12:00:00', 'severity': 'high', ...}

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

def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary for logging and serialization.

    Returns:
        Flattened dictionary with namespaced keys.

    Examples:
        >>> ctx.to_dict()
        {'timestamp': '2024-01-01T12:00:00', 'severity': 'high', ...}

    """
    result: dict[str, Any] = {
        "timestamp": self.timestamp.isoformat(),
        "severity": self.severity.value,
        "category": self.category.value,
    }

    # Add tracing if present
    if self.trace_id:
        result["trace_id"] = self.trace_id
    if self.span_id:
        result["span_id"] = self.span_id

    # Flatten namespaced metadata
    for namespace, data in self.metadata.items():
        for key, value in data.items():
            result[f"{namespace}.{key}"] = value

    # Add tags if present
    if self.tags:
        result["tags"] = sorted(list(self.tags))

    return result

to_logging_context

to_logging_context() -> dict[str, Any]

Convert to context suitable for structured logging.

Returns:

Type	Description
dict[str, Any]	Dictionary formatted for logger context.

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

def to_logging_context(self) -> dict[str, Any]:
    """Convert to context suitable for structured logging.

    Returns:
        Dictionary formatted for logger context.

    """
    return self.to_dict()

to_terraform_diagnostic

to_terraform_diagnostic() -> dict[str, Any]

Convert to Terraform diagnostic format.

Returns:

Type	Description
dict[str, Any]	Dictionary formatted for Terraform diagnostics.

Examples:

>>> ctx.to_terraform_diagnostic()
{'severity': 'error', 'summary': '...', 'detail': {...}}

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

def to_terraform_diagnostic(self) -> dict[str, Any]:
    """Convert to Terraform diagnostic format.

    Returns:
        Dictionary formatted for Terraform diagnostics.

    Examples:
        >>> ctx.to_terraform_diagnostic()
        {'severity': 'error', 'summary': '...', 'detail': {...}}

    """
    # Map our severity to Terraform severity
    severity_map = {
        ErrorSeverity.LOW: "warning",
        ErrorSeverity.MEDIUM: "warning",
        ErrorSeverity.HIGH: "error",
        ErrorSeverity.CRITICAL: "error",
    }

    diagnostic: dict[str, Any] = {
        "severity": severity_map[self.severity],
        "detail": {},
    }

    # Add Terraform-specific metadata if present
    tf_meta = self.get_namespace("terraform")
    if tf_meta:
        diagnostic["detail"].update(tf_meta)

    # Add other relevant metadata
    for namespace, data in self.metadata.items():
        if namespace != "terraform":
            diagnostic["detail"][namespace] = data

    return diagnostic

update_namespace

update_namespace(
    namespace: str, data: dict[str, Any]
) -> ErrorContext

Update existing namespace metadata.

Parameters:

Name	Type	Description	Default
namespace	str	Namespace key to update.	required
data	dict[str, Any]	Metadata to merge into namespace.	required

Returns:

Type	Description
ErrorContext	Self for method chaining.

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

def update_namespace(self, namespace: str, data: dict[str, Any]) -> ErrorContext:
    """Update existing namespace metadata.

    Args:
        namespace: Namespace key to update.
        data: Metadata to merge into namespace.

    Returns:
        Self for method chaining.

    """
    if namespace not in self.metadata:
        self.metadata[namespace] = {}
    self.metadata[namespace].update(data)
    return self

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

ErrorMetadata

Additional metadata for error tracking and debugging.

Attributes:

Name	Type	Description
request_id	str | None	Optional request identifier for tracing.
user_id	str | None	Optional user identifier.
session_id	str | None	Optional session identifier.
correlation_id	str | None	Optional correlation ID for distributed tracing.
retry_count	int	Number of retry attempts made.
retry_after	float | None	Seconds to wait before retry.
help_url	str | None	Optional URL for more information.
support_id	str | None	Optional support ticket/case ID.

Examples:

>>> meta = ErrorMetadata(
...     request_id="req_123",
...     user_id="user_456",
...     retry_count=2
... )

Functions

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary, excluding None values.

Returns:

Type	Description
dict[str, Any]	Dictionary with non-None metadata fields.

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

def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary, excluding None values.

    Returns:
        Dictionary with non-None metadata fields.

    """
    result = {}
    for key in [
        "request_id",
        "user_id",
        "session_id",
        "correlation_id",
        "retry_count",
        "retry_after",
        "help_url",
        "support_id",
    ]:
        value = getattr(self, key)
        if value is not None:
            result[key] = value
    return result

ErrorSeverity

Bases: str, Enum

Error severity levels for prioritization and alerting.

FoundationError

FoundationError(
    message: str,
    *,
    code: str | None = None,
    context: dict[str, Any] | None = None,
    cause: Exception | None = None,
    **extra_context: Any
)

Bases: Exception

Base exception for all Foundation errors.

Parameters:

Name	Type	Description	Default
message	str	Human-readable error message.	required
code	str | None	Optional error code for programmatic handling.	None
context	dict[str, Any] | None	Optional context dictionary with diagnostic data.	None
cause	Exception | None	Optional underlying exception that caused this error.	None
**extra_context	Any	Additional key-value pairs added to context.	{}

Examples:

>>> raise FoundationError("Operation failed")
>>> raise FoundationError("Operation failed", code="OP_001")
>>> raise FoundationError("Operation failed", user_id=123, retry_count=3)

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

def __init__(
    self,
    message: str,
    *,
    code: str | None = None,
    context: dict[str, Any] | None = None,
    cause: Exception | None = None,
    **extra_context: Any,
) -> None:
    self.message = message
    self.code = code or self._default_code()
    self.context = context or {}
    self.context.update(extra_context)
    self.cause = cause
    if cause:
        self.__cause__ = cause
    super().__init__(message)

Functions

add_context

add_context(key: str, value: Any) -> FoundationError

Add context data to the error.

Parameters:

Name	Type	Description	Default
key	str	Context key (use dots for namespacing, e.g., 'aws.region').	required
value	Any	Context value.	required

Returns:

Type	Description
FoundationError	Self for method chaining.

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

def add_context(self, key: str, value: Any) -> FoundationError:
    """Add context data to the error.

    Args:
        key: Context key (use dots for namespacing, e.g., 'aws.region').
        value: Context value.

    Returns:
        Self for method chaining.

    """
    self.context[key] = value
    return self

to_dict

to_dict() -> dict[str, Any]

Convert exception to dictionary for structured logging.

Returns:

Type	Description
dict[str, Any]	Dictionary representation suitable for logging/serialization.

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

def to_dict(self) -> dict[str, Any]:
    """Convert exception to dictionary for structured logging.

    Returns:
        Dictionary representation suitable for logging/serialization.

    """
    result = {
        "error.type": self.__class__.__name__,
        "error.message": self.message,
        "error.code": self.code,
    }

    # Add context with error prefix
    for key, value in self.context.items():
        # If key already has a prefix, use it; otherwise add error prefix
        if "." in key:
            result[key] = value
        else:
            result[f"error.{key}"] = value

    if self.cause:
        result["error.cause"] = str(self.cause)
        result["error.cause_type"] = type(self.cause).__name__

    return result

IntegrationError

IntegrationError(
    message: str,
    *,
    service: str | None = None,
    endpoint: str | None = None,
    status_code: int | None = None,
    **kwargs: Any
)

Bases: FoundationError

Raised when external service integration fails.

Parameters:

Name	Type	Description	Default
message	str	Error message describing the integration failure.	required
service	str | None	Optional service name that failed.	None
endpoint	str | None	Optional endpoint that was called.	None
status_code	int | None	Optional HTTP status code.	None
**kwargs	Any	Additional context passed to FoundationError.	{}

Examples:

>>> raise IntegrationError("API call failed")
>>> raise IntegrationError("Auth failed", service="github", status_code=401)

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

def __init__(
    self,
    message: str,
    *,
    service: str | None = None,
    endpoint: str | None = None,
    status_code: int | None = None,
    **kwargs: Any,
) -> None:
    if service:
        kwargs.setdefault("context", {})["integration.service"] = service
    if endpoint:
        kwargs.setdefault("context", {})["integration.endpoint"] = endpoint
    if status_code:
        kwargs.setdefault("context", {})["integration.status_code"] = status_code
    super().__init__(message, **kwargs)

NetworkError

NetworkError(
    message: str,
    *,
    host: str | None = None,
    port: int | None = None,
    **kwargs: Any
)

Bases: IntegrationError

Raised for network-related failures.

Parameters:

Name	Type	Description	Default
message	str	Error message describing the network issue.	required
host	str | None	Optional hostname or IP address.	None
port	int | None	Optional port number.	None
**kwargs	Any	Additional context passed to IntegrationError.	{}

Examples:

>>> raise NetworkError("Connection refused")
>>> raise NetworkError("DNS resolution failed", host="api.example.com")

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

def __init__(
    self,
    message: str,
    *,
    host: str | None = None,
    port: int | None = None,
    **kwargs: Any,
) -> None:
    if host:
        kwargs.setdefault("context", {})["network.host"] = host
    if port:
        kwargs.setdefault("context", {})["network.port"] = port
    super().__init__(message, **kwargs)

NotFoundError

NotFoundError(
    message: str,
    *,
    resource_type: str | None = None,
    resource_id: str | None = None,
    **kwargs: Any
)

Bases: FoundationError

Raised when a requested resource cannot be found.

Parameters:

Name	Type	Description	Default
message	str	Error message describing what was not found.	required
resource_type	str | None	Optional type of resource.	None
resource_id	str | None	Optional resource identifier.	None
**kwargs	Any	Additional context passed to FoundationError.	{}

Examples:

>>> raise NotFoundError("User not found")
>>> raise NotFoundError("Entity missing", resource_type="user", resource_id="123")

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

def __init__(
    self,
    message: str,
    *,
    resource_type: str | None = None,
    resource_id: str | None = None,
    **kwargs: Any,
) -> None:
    if resource_type:
        kwargs.setdefault("context", {})["notfound.type"] = resource_type
    if resource_id:
        kwargs.setdefault("context", {})["notfound.id"] = resource_id
    super().__init__(message, **kwargs)

ProcessError

ProcessError(
    message: str,
    *,
    command: str | list[str] | None = None,
    return_code: int | None = None,
    stdout: str | bytes | None = None,
    stderr: str | bytes | None = None,
    timeout: bool = False,
    code: str | None = None,
    **extra_context: Any
)

Bases: FoundationError

Error for external process execution failures with output capture.

Initialize ProcessError with command execution details.

Parameters:

Name	Type	Description	Default
message	str	Human-readable error message	required
command	str | list[str] | None	The command that was executed	None
return_code	int | None	Process return/exit code	None
stdout	str | bytes | None	Standard output from the process	None
stderr	str | bytes | None	Standard error from the process	None
timeout	bool	Whether the process timed out	False
code	str | None	Optional error code	None
**extra_context	Any	Additional context information	{}

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

def __init__(
    self,
    message: str,
    *,
    command: str | list[str] | None = None,
    return_code: int | None = None,
    stdout: str | bytes | None = None,
    stderr: str | bytes | None = None,
    timeout: bool = False,
    code: str | None = None,
    **extra_context: Any,
) -> None:
    """Initialize ProcessError with command execution details.

    Args:
        message: Human-readable error message
        command: The command that was executed
        return_code: Process return/exit code
        stdout: Standard output from the process
        stderr: Standard error from the process
        timeout: Whether the process timed out
        code: Optional error code
        **extra_context: Additional context information

    """
    # Build comprehensive error message
    full_message = message

    if command:
        cmd_str = command if isinstance(command, str) else " ".join(command)
        full_message += f"\nCommand: {cmd_str}"

    if return_code is not None:
        full_message += f"\nReturn code: {return_code}"

    if timeout:
        full_message += "\nProcess timed out"

    if stdout:
        stdout_str = stdout.decode("utf-8", "replace") if isinstance(stdout, bytes) else stdout
        if stdout_str.strip():
            full_message += f"\n--- STDOUT ---\n{stdout_str.strip()}"

    if stderr:
        stderr_str = stderr.decode("utf-8", "replace") if isinstance(stderr, bytes) else stderr
        if stderr_str.strip():
            full_message += f"\n--- STDERR ---\n{stderr_str.strip()}"

    # Store structured data
    context = extra_context.copy()
    context.update(
        {
            "process.command": command,
            "process.return_code": return_code,
            "process.timeout": timeout,
        },
    )

    # Store clean stdout/stderr for programmatic access
    self.stdout = (
        stdout.decode("utf-8", "replace").strip()
        if isinstance(stdout, bytes)
        else stdout.strip()
        if stdout
        else None
    )

    self.stderr = (
        stderr.decode("utf-8", "replace").strip()
        if isinstance(stderr, bytes)
        else stderr.strip()
        if stderr
        else None
    )

    self.command = command
    self.return_code = return_code
    self.timeout = timeout

    super().__init__(full_message, code=code, context=context)

Functions

ProcessTimeoutError

ProcessTimeoutError(
    message: str,
    *,
    command: str | list[str] | None = None,
    timeout_seconds: float | None = None,
    stdout: str | bytes | None = None,
    stderr: str | bytes | None = None,
    code: str | None = None,
    **extra_context: Any
)

Bases: ProcessError

Error when a process times out.

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

def __init__(
    self,
    message: str,
    *,
    command: str | list[str] | None = None,
    timeout_seconds: float | None = None,
    stdout: str | bytes | None = None,
    stderr: str | bytes | None = None,
    code: str | None = None,
    **extra_context: Any,
) -> None:
    context = extra_context.copy()
    if timeout_seconds is not None:
        context["process.timeout_seconds"] = timeout_seconds

    super().__init__(
        message,
        command=command,
        stdout=stdout,
        stderr=stderr,
        timeout=True,
        code=code,
        **context,
    )

RateLimitExceededError

RateLimitExceededError(
    message: str,
    *,
    limit: float | None = None,
    retry_after: float | None = None,
    current_rate: float | None = None,
    **kwargs: Any
)

Bases: FoundationError

Raised when a rate limit is exceeded.

Parameters:

Name	Type	Description	Default
message	str	Error message describing the rate limit violation.	required
limit	float | None	The rate limit that was exceeded (requests/messages per time unit).	None
retry_after	float | None	Seconds to wait before retrying.	None
current_rate	float | None	Optional current rate at time of error.	None
**kwargs	Any	Additional context passed to FoundationError.	{}

Examples:

>>> raise RateLimitExceededError("Log rate limit exceeded", limit=100.0, retry_after=1.0)
>>> raise RateLimitExceededError("API rate limit", limit=1000, retry_after=60, current_rate=1050)

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

def __init__(
    self,
    message: str,
    *,
    limit: float | None = None,
    retry_after: float | None = None,
    current_rate: float | None = None,
    **kwargs: Any,
) -> None:
    if limit is not None:
        kwargs.setdefault("context", {})["rate_limit.limit"] = limit
    if retry_after is not None:
        kwargs.setdefault("context", {})["rate_limit.retry_after"] = retry_after
    if current_rate is not None:
        kwargs.setdefault("context", {})["rate_limit.current_rate"] = current_rate
    super().__init__(message, **kwargs)

ResourceError

ResourceError(
    message: str,
    *,
    resource_type: str | None = None,
    resource_path: str | None = None,
    **kwargs: Any
)

Bases: FoundationError

Raised when resource operations fail.

Parameters:

Name	Type	Description	Default
message	str	Error message describing the resource issue.	required
resource_type	str | None	Optional type of resource (file, network, etc.).	None
resource_path	str | None	Optional path or identifier of the resource.	None
**kwargs	Any	Additional context passed to FoundationError.	{}

Examples:

>>> raise ResourceError("File not found")
>>> raise ResourceError("Permission denied", resource_type="file", resource_path="/etc/config")

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

def __init__(
    self,
    message: str,
    *,
    resource_type: str | None = None,
    resource_path: str | None = None,
    **kwargs: Any,
) -> None:
    if resource_type:
        kwargs.setdefault("context", {})["resource.type"] = resource_type
    if resource_path:
        kwargs.setdefault("context", {})["resource.path"] = resource_path
    super().__init__(message, **kwargs)

RuntimeError

RuntimeError(
    message: str,
    *,
    operation: str | None = None,
    retry_possible: bool = False,
    **kwargs: Any
)

Bases: FoundationError

Raised for runtime operational errors.

Parameters:

Name	Type	Description	Default
message	str	Error message describing the runtime issue.	required
operation	str | None	Optional operation that failed.	None
retry_possible	bool	Whether the operation can be retried.	False
**kwargs	Any	Additional context passed to FoundationError.	{}

Examples:

>>> raise RuntimeError("Process failed")
>>> raise RuntimeError("Lock timeout", operation="acquire_lock", retry_possible=True)

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

def __init__(
    self,
    message: str,
    *,
    operation: str | None = None,
    retry_possible: bool = False,
    **kwargs: Any,
) -> None:
    if operation:
        kwargs.setdefault("context", {})["runtime.operation"] = operation
    kwargs.setdefault("context", {})["runtime.retry_possible"] = retry_possible
    super().__init__(message, **kwargs)

StateError

StateError(
    message: str,
    *,
    current_state: str | None = None,
    expected_state: str | None = None,
    transition: str | None = None,
    **kwargs: Any
)

Bases: FoundationError

Raised when an operation is invalid for the current state.

Parameters:

Name	Type	Description	Default
message	str	Error message describing the state issue.	required
current_state	str | None	Optional current state.	None
expected_state	str | None	Optional expected state.	None
transition	str | None	Optional attempted transition.	None
**kwargs	Any	Additional context passed to FoundationError.	{}

Examples:

>>> raise StateError("Invalid state transition")
>>> raise StateError("Not ready", current_state="initializing", expected_state="ready")

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

def __init__(
    self,
    message: str,
    *,
    current_state: str | None = None,
    expected_state: str | None = None,
    transition: str | None = None,
    **kwargs: Any,
) -> None:
    if current_state:
        kwargs.setdefault("context", {})["state.current"] = current_state
    if expected_state:
        kwargs.setdefault("context", {})["state.expected"] = expected_state
    if transition:
        kwargs.setdefault("context", {})["state.transition"] = transition
    super().__init__(message, **kwargs)

TimeoutError

TimeoutError(
    message: str,
    *,
    timeout_seconds: float | None = None,
    elapsed_seconds: float | None = None,
    **kwargs: Any
)

Bases: IntegrationError

Raised when operations exceed time limits.

Parameters:

Name	Type	Description	Default
message	str	Error message describing the timeout.	required
timeout_seconds	float | None	Optional timeout limit in seconds.	None
elapsed_seconds	float | None	Optional actual elapsed time.	None
**kwargs	Any	Additional context passed to IntegrationError.	{}

Examples:

>>> raise TimeoutError("Request timed out")
>>> raise TimeoutError("Operation exceeded limit", timeout_seconds=30, elapsed_seconds=31.5)

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

def __init__(
    self,
    message: str,
    *,
    timeout_seconds: float | None = None,
    elapsed_seconds: float | None = None,
    **kwargs: Any,
) -> None:
    if timeout_seconds is not None:
        kwargs.setdefault("context", {})["timeout.limit"] = timeout_seconds
    if elapsed_seconds is not None:
        kwargs.setdefault("context", {})["timeout.elapsed"] = elapsed_seconds
    super().__init__(message, **kwargs)

ValidationError

ValidationError(
    message: str,
    *,
    field: str | None = None,
    value: Any = None,
    rule: str | None = None,
    **kwargs: Any
)

Bases: FoundationError

Raised when data validation fails.

Parameters:

Name	Type	Description	Default
message	str	Validation error message.	required
field	str | None	Optional field name that failed validation.	None
value	Any	Optional invalid value.	None
rule	str | None	Optional validation rule that failed.	None
**kwargs	Any	Additional context passed to FoundationError.	{}

Examples:

>>> raise ValidationError("Invalid email format")
>>> raise ValidationError("Value out of range", field="age", value=-1)

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

def __init__(
    self,
    message: str,
    *,
    field: str | None = None,
    value: Any = None,
    rule: str | None = None,
    **kwargs: Any,
) -> None:
    if field:
        kwargs.setdefault("context", {})["validation.field"] = field
    if value is not None:
        kwargs.setdefault("context", {})["validation.value"] = str(value)
    if rule:
        kwargs.setdefault("context", {})["validation.rule"] = rule
    super().__init__(message, **kwargs)

Functions

capture_error_context

capture_error_context(
    error: Exception,
    severity: ErrorSeverity | None = None,
    category: ErrorCategory | None = None,
    **namespaces: dict[str, Any]
) -> ErrorContext

Capture error context from an exception.

Parameters:

Name	Type	Description	Default
error	Exception	Exception to capture context from.	required
severity	ErrorSeverity | None	Optional severity override.	None
category	ErrorCategory | None	Optional category override.	None
**namespaces	dict[str, Any]	Namespace data to add to context.	{}

Returns:

Type	Description
ErrorContext	ErrorContext with captured information.

Examples:

>>> try:
...     risky_operation()
... except Exception as e:
...     ctx = capture_error_context(
...         e,
...         severity=ErrorSeverity.HIGH,
...         aws={"region": "us-east-1"},
...         http={"status": 500}
...     )

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

def capture_error_context(
    error: Exception,
    severity: ErrorSeverity | None = None,
    category: ErrorCategory | None = None,
    **namespaces: dict[str, Any],
) -> ErrorContext:
    """Capture error context from an exception.

    Args:
        error: Exception to capture context from.
        severity: Optional severity override.
        category: Optional category override.
        **namespaces: Namespace data to add to context.

    Returns:
        ErrorContext with captured information.

    Examples:
        >>> try:
        ...     risky_operation()
        ... except Exception as e:
        ...     ctx = capture_error_context(
        ...         e,
        ...         severity=ErrorSeverity.HIGH,
        ...         aws={"region": "us-east-1"},
        ...         http={"status": 500}
        ...     )

    """
    # Determine severity and category if not provided
    if severity is None:
        severity = _determine_error_severity(error)
    if category is None:
        category = _determine_error_category(error)

    # Create context
    ctx = ErrorContext(severity=severity, category=category)

    # Add error details
    ctx.add_namespace(
        "error",
        {
            "type": type(error).__name__,
            "message": str(error),
        },
    )

    # Add any namespaces provided
    for namespace, data in namespaces.items():
        ctx.add_namespace(namespace, data)

    # Add context from FoundationError if applicable
    from provide.foundation.errors.base import FoundationError

    if isinstance(error, FoundationError) and error.context:
        grouped = _group_foundation_error_context(error.context)
        for namespace, data in grouped.items():
            ctx.update_namespace(namespace, data)

    return ctx

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

fallback_on_error

fallback_on_error(
    fallback_func: Callable[..., Any],
    *exceptions: type[Exception],
    log_errors: bool = True
) -> Callable[[F], F]

Decorator to call a fallback function when errors occur.

Parameters:

Name	Type	Description	Default
fallback_func	Callable[..., Any]	Function to call when an error occurs.	required
*exceptions	type[Exception]	Specific exception types to handle (all if empty).	()
log_errors	bool	Whether to log errors before calling fallback.	True

Returns:

Type	Description
Callable[[F], F]	Decorated function.

Examples:

>>> def use_cache():
...     return cached_value
...
>>> @fallback_on_error(use_cache, NetworkError)
... def fetch_from_api():
...     return api_call()

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

def fallback_on_error(
    fallback_func: Callable[..., Any],
    *exceptions: type[Exception],
    log_errors: bool = True,
) -> Callable[[F], F]:
    """Decorator to call a fallback function when errors occur.

    Args:
        fallback_func: Function to call when an error occurs.
        *exceptions: Specific exception types to handle (all if empty).
        log_errors: Whether to log errors before calling fallback.

    Returns:
        Decorated function.

    Examples:
        >>> def use_cache():
        ...     return cached_value
        ...
        >>> @fallback_on_error(use_cache, NetworkError)
        ... def fetch_from_api():
        ...     return api_call()

    """
    catch_types = exceptions if exceptions else (Exception,)

    def decorator(func: F) -> F:
        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            try:
                return func(*args, **kwargs)
            except catch_types as e:
                if log_errors:
                    from provide.foundation.hub.foundation import get_foundation_logger

                    get_foundation_logger().warning(
                        f"Using fallback for {getattr(func, '__name__', '<anonymous>')} due to {type(e).__name__}",
                        function=getattr(func, "__name__", "<anonymous>"),
                        error_type=type(e).__name__,
                        error=str(e),
                        fallback=getattr(fallback_func, "__name__", "<anonymous>"),
                    )

                # Call fallback with same arguments
                try:
                    return fallback_func(*args, **kwargs)
                except Exception as fallback_error:
                    from provide.foundation.hub.foundation import get_foundation_logger

                    get_foundation_logger().error(
                        f"Fallback function {getattr(fallback_func, '__name__', '<anonymous>')} also failed",
                        exc_info=True,
                        original_error=str(e),
                        fallback_error=str(fallback_error),
                    )
                    # Re-raise the fallback error
                    raise fallback_error from e

        return wrapper  # type: ignore

    return decorator

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

log_only_error_context

log_only_error_context(
    *,
    context_provider: (
        Callable[[], dict[str, Any]] | None
    ) = None,
    log_level: str = "debug",
    log_success: bool = False
) -> Callable[[F], F]

Safe decorator that only adds logging context without changing error behavior.

This decorator preserves the exact original error message and type while adding structured logging context. It never suppresses errors or changes their behavior.

Parameters:

Name	Type	Description	Default
context_provider	Callable[[], dict[str, Any]] | None	Function that provides additional logging context.	None
log_level	str	Level for operation logging ('debug', 'trace', etc.)	'debug'
log_success	bool	Whether to log successful operations.	False

Returns:

Type	Description
Callable[[F], F]	Decorated function that preserves all original error behavior.

Examples:

>>> @log_only_error_context(
...     context_provider=lambda: {"operation": "detect_launcher_type"},
...     log_level="trace"
... )
... def detect_launcher_type(self, path):
...     # Original error messages preserved exactly
...     return self._internal_detect(path)

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

def log_only_error_context(
    *,
    context_provider: Callable[[], dict[str, Any]] | None = None,
    log_level: str = "debug",
    log_success: bool = False,
) -> Callable[[F], F]:
    """Safe decorator that only adds logging context without changing error behavior.

    This decorator preserves the exact original error message and type while adding
    structured logging context. It never suppresses errors or changes their behavior.

    Args:
        context_provider: Function that provides additional logging context.
        log_level: Level for operation logging ('debug', 'trace', etc.)
        log_success: Whether to log successful operations.

    Returns:
        Decorated function that preserves all original error behavior.

    Examples:
        >>> @log_only_error_context(
        ...     context_provider=lambda: {"operation": "detect_launcher_type"},
        ...     log_level="trace"
        ... )
        ... def detect_launcher_type(self, path):
        ...     # Original error messages preserved exactly
        ...     return self._internal_detect(path)

    """

    def decorator(func: F) -> F:
        if inspect.iscoroutinefunction(func):

            @functools.wraps(func)
            async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                context = context_provider() if context_provider else {}
                logger = get_foundation_logger()

                _log_function_entry(logger, func, log_level, context)

                try:
                    result = await func(*args, **kwargs)

                    if log_success:
                        _log_function_success(logger, func, log_level, context)

                    return result

                except Exception as e:
                    _log_function_error(logger, func, e, context)
                    raise

            return async_wrapper  # type: ignore

        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            context = context_provider() if context_provider else {}
            logger = get_foundation_logger()

            _log_function_entry(logger, func, log_level, context)

            try:
                result = func(*args, **kwargs)

                if log_success:
                    _log_function_success(logger, func, log_level, context)

                return result

            except Exception as e:
                _log_function_error(logger, func, e, context)
                raise

        return wrapper  # type: ignore

    return decorator

resilient

resilient(func: F) -> F

resilient(
    func: None = None,
    *,
    fallback: Any = None,
    log_errors: bool = True,
    context_provider: (
        Callable[[], dict[str, Any]] | None
    ) = None,
    context: dict[str, Any] | None = None,
    error_mapper: (
        Callable[[Exception], Exception] | None
    ) = None,
    suppress: tuple[type[Exception], ...] | None = None,
    reraise: bool = True
) -> Callable[[F], F]

resilient(
    func: F | None = None,
    *,
    fallback: Any = None,
    log_errors: bool = True,
    context_provider: (
        Callable[[], dict[str, Any]] | None
    ) = None,
    context: dict[str, Any] | None = None,
    error_mapper: (
        Callable[[Exception], Exception] | None
    ) = None,
    suppress: tuple[type[Exception], ...] | None = None,
    reraise: bool = True
) -> Callable[[F], F] | F

Decorator for automatic error handling with logging.

Parameters:

Name	Type	Description	Default
fallback	Any	Value to return when an error occurs.	None
log_errors	bool	Whether to log errors.	True
context_provider	Callable[[], dict[str, Any]] | None	Function that provides additional logging context.	None
context	dict[str, Any] | None	Static context dict to include in logs (alternative to context_provider).	None
error_mapper	Callable[[Exception], Exception] | None	Function to transform exceptions before re-raising.	None
suppress	tuple[type[Exception], ...] | None	Tuple of exception types to suppress (return fallback instead).	None
reraise	bool	Whether to re-raise exceptions after logging (default: True).	True

Returns:

Type	Description
Callable[[F], F] | F	Decorated function.

Note

Preserving Context in error_mapper: When using error_mapper with FoundationError exceptions, the original exception's context dictionary is not automatically transferred to the mapped exception. To preserve rich context, manually copy it:

from provide.foundation.errors import FoundationError @resilient( ... error_mapper=lambda e: ( ... ValidationError( ... str(e), ... context=e.context if isinstance(e, FoundationError) else {} ... ) if isinstance(e, FoundationError) ... else DomainError(str(e)) ... ) ... ) ... def process_data(data): ... # Low-level FoundationError will be mapped to ValidationError ... # with context preserved ... pass

Examples:

>>> @resilient(fallback=None, suppress=(KeyError,))
... def get_value(data, key):
...     return data[key]

>>> @resilient(
...     context_provider=lambda: {"request_id": get_request_id()}
... )
... def process_request():
...     # errors will be logged with request_id
...     pass

>>> @resilient(
...     reraise=False,
...     context={"component": "orchestrator", "method": "run"}
... )
... def run():
...     # errors will be logged but not re-raised
...     pass

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

def resilient(
    func: F | None = None,
    *,
    fallback: Any = None,
    log_errors: bool = True,
    context_provider: Callable[[], dict[str, Any]] | None = None,
    context: dict[str, Any] | None = None,
    error_mapper: Callable[[Exception], Exception] | None = None,
    suppress: tuple[type[Exception], ...] | None = None,
    reraise: bool = True,
) -> Callable[[F], F] | F:
    """Decorator for automatic error handling with logging.

    Args:
        fallback: Value to return when an error occurs.
        log_errors: Whether to log errors.
        context_provider: Function that provides additional logging context.
        context: Static context dict to include in logs (alternative to context_provider).
        error_mapper: Function to transform exceptions before re-raising.
        suppress: Tuple of exception types to suppress (return fallback instead).
        reraise: Whether to re-raise exceptions after logging (default: True).

    Returns:
        Decorated function.

    Note:
        **Preserving Context in error_mapper:**
        When using error_mapper with FoundationError exceptions, the original
        exception's context dictionary is not automatically transferred to the
        mapped exception. To preserve rich context, manually copy it:

        >>> from provide.foundation.errors import FoundationError
        >>> @resilient(
        ...     error_mapper=lambda e: (
        ...         ValidationError(
        ...             str(e),
        ...             context=e.context if isinstance(e, FoundationError) else {}
        ...         ) if isinstance(e, FoundationError)
        ...         else DomainError(str(e))
        ...     )
        ... )
        ... def process_data(data):
        ...     # Low-level FoundationError will be mapped to ValidationError
        ...     # with context preserved
        ...     pass

    Examples:
        >>> @resilient(fallback=None, suppress=(KeyError,))
        ... def get_value(data, key):
        ...     return data[key]

        >>> @resilient(
        ...     context_provider=lambda: {"request_id": get_request_id()}
        ... )
        ... def process_request():
        ...     # errors will be logged with request_id
        ...     pass

        >>> @resilient(
        ...     reraise=False,
        ...     context={"component": "orchestrator", "method": "run"}
        ... )
        ... def run():
        ...     # errors will be logged but not re-raised
        ...     pass

    """

    def decorator(func: F) -> F:
        # Create error handler with all configuration
        handler = ResilientErrorHandler(
            fallback=fallback,
            log_errors=log_errors,
            context_provider=context_provider,
            context=context,
            error_mapper=error_mapper,
            suppress=suppress,
            reraise=reraise,
        )

        # Return appropriate wrapper based on function type
        if inspect.iscoroutinefunction(func):
            return _create_async_wrapper(func, handler)
        return _create_sync_wrapper(func, handler)

    # Support both @resilient and @resilient(...) forms
    if func is None:
        return decorator
    return decorator(func)

suppress_and_log

suppress_and_log(
    *exceptions: type[Exception],
    fallback: Any = None,
    log_level: str = "warning"
) -> Callable[[F], F]

Decorator to suppress specific exceptions and log them.

Parameters:

Name	Type	Description	Default
*exceptions	type[Exception]	Exception types to suppress.	()
fallback	Any	Value to return when exception is suppressed.	None
log_level	str	Log level to use ('debug', 'info', 'warning', 'error').	'warning'

Returns:

Type	Description
Callable[[F], F]	Decorated function.

Examples:

>>> @suppress_and_log(KeyError, AttributeError, fallback={})
... def get_nested_value(data):
...     return data["key"].attribute

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

def suppress_and_log(
    *exceptions: type[Exception],
    fallback: Any = None,
    log_level: str = "warning",
) -> Callable[[F], F]:
    """Decorator to suppress specific exceptions and log them.

    Args:
        *exceptions: Exception types to suppress.
        fallback: Value to return when exception is suppressed.
        log_level: Log level to use ('debug', 'info', 'warning', 'error').

    Returns:
        Decorated function.

    Examples:
        >>> @suppress_and_log(KeyError, AttributeError, fallback={})
        ... def get_nested_value(data):
        ...     return data["key"].attribute

    """

    def decorator(func: F) -> F:
        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            try:
                return func(*args, **kwargs)
            except exceptions as e:
                # Get appropriate log method
                from provide.foundation.hub.foundation import get_foundation_logger

                if log_level in ("debug", "info", "warning", "error", "critical"):
                    log_method = getattr(get_foundation_logger(), log_level)
                else:
                    log_method = get_foundation_logger().warning

                log_method(
                    f"Suppressed {type(e).__name__} in {getattr(func, '__name__', '<anonymous>')}: {e}",
                    function=getattr(func, "__name__", "<anonymous>"),
                    error_type=type(e).__name__,
                    error=str(e),
                    fallback=fallback,
                )

                return fallback

        return wrapper  # type: ignore

    return decorator

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