Skip to content

softfail

azure_bootstrap.softfail

Helpers for the "log + alert + continue with degraded result" pattern.

Apps wrap fragile code paths (AI summarization, optional enrichment, etc.) in :func:soft_fail_with or the :func:soft_fail context manager so a single sub-feature failure produces a degraded result rather than killing the whole pipeline.

By default, both helpers re-raise exceptions that are unrecoverable per :func:azure_bootstrap.exceptions.is_unrecoverable — soft-failing an InvalidMessageError would break the dead-letter contract.

Functions:

Name Description
soft_fail_with

Call fn(*args, **kwargs); soft-fail with fallback on caught error.
soft_fail

Context-manager form of :func:soft_fail_with.

soft_fail_with 

soft_fail_with(fn: Callable[..., T], *args: Any, fallback: T, catch: type[BaseException] | tuple[type[BaseException], ...] = Exception, operation: str, alert_severity: str | None = 'error', counter_name: str | None = None, fallback_fn: Callable[[BaseException], T] | None = None, re_raise_unrecoverable: bool = True, **kwargs: Any) -> SoftFailResult[T]

Call fn(*args, **kwargs); soft-fail with fallback on caught error.

Unrecoverable exceptions (per :func:is_unrecoverable) propagate by default — set re_raise_unrecoverable=False to explicitly suppress them too.

Source code in azure_bootstrap/softfail/__init__.py 
54
55
56
57
58
59
60
61
62
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
90
91
92
93
94
95
96
97
98
99
def soft_fail_with(
    fn: Callable[..., T],
    *args: Any,
    fallback: T,
    catch: type[BaseException] | tuple[type[BaseException], ...] = Exception,
    operation: str,
    alert_severity: str | None = "error",
    counter_name: str | None = None,
    fallback_fn: Callable[[BaseException], T] | None = None,
    re_raise_unrecoverable: bool = True,
    **kwargs: Any,
) -> SoftFailResult[T]:
    """Call ``fn(*args, **kwargs)``; soft-fail with ``fallback`` on caught error.

    Unrecoverable exceptions (per :func:`is_unrecoverable`) propagate by
    default — set ``re_raise_unrecoverable=False`` to explicitly suppress
    them too.
    """
    try:
        value = fn(*args, **kwargs)
    except catch as exc:
        if re_raise_unrecoverable and is_unrecoverable(exc):
            raise
        _logger.warning(
            "soft-fail in %s: %s",
            operation,
            type(exc).__name__,
            exc_info=True,
            extra={
                "operation": operation,
                "exception_type": type(exc).__name__,
                "error": str(exc)[:500],
            },
        )
        if counter_name:
            bump_counter(counter_name)
        if alert_severity:
            _fire_alert(alert_severity, operation, exc)
        resolved = fallback_fn(exc) if fallback_fn else fallback
        return SoftFailResult(
            value=resolved,
            degraded=True,
            reason=type(exc).__name__,
            exception=exc,
        )
    return SoftFailResult(value=value, degraded=False)

soft_fail 

soft_fail(*, operation: str, catch: type[BaseException] | tuple[type[BaseException], ...] = Exception, alert_severity: str | None = 'error', counter_name: str | None = None, re_raise_unrecoverable: bool = True) -> Generator[dict[str, Any], None, None]

Context-manager form of :func:soft_fail_with.

Yields a mutable dict the caller can inspect after the block::

with soft_fail(operation='ai.summary', counter_name='ai.summary.failed') as ctx:
    summary = ai.summarize(text)
if ctx['degraded']:
    summary = None  # caller handles the degraded case
Source code in azure_bootstrap/softfail/__init__.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
@contextmanager
def soft_fail(
    *,
    operation: str,
    catch: type[BaseException] | tuple[type[BaseException], ...] = Exception,
    alert_severity: str | None = "error",
    counter_name: str | None = None,
    re_raise_unrecoverable: bool = True,
) -> Generator[dict[str, Any], None, None]:
    """Context-manager form of :func:`soft_fail_with`.

    Yields a mutable dict the caller can inspect after the block::

        with soft_fail(operation='ai.summary', counter_name='ai.summary.failed') as ctx:
            summary = ai.summarize(text)
        if ctx['degraded']:
            summary = None  # caller handles the degraded case
    """
    state: dict[str, Any] = {"degraded": False, "reason": None, "exception": None}
    try:
        yield state
    except catch as exc:
        if re_raise_unrecoverable and is_unrecoverable(exc):
            raise
        state["degraded"] = True
        state["reason"] = type(exc).__name__
        state["exception"] = exc
        _logger.warning(
            "soft-fail in %s: %s",
            operation,
            type(exc).__name__,
            exc_info=True,
            extra={
                "operation": operation,
                "exception_type": type(exc).__name__,
                "error": str(exc)[:500],
            },
        )
        if counter_name:
            bump_counter(counter_name)
        if alert_severity:
            _fire_alert(alert_severity, operation, exc)