Skip to content

phases

azure_bootstrap.phases

Per-phase guarded execution for multi-stage pipelines.

The point: one phase's bug must not nuke the whole run. run_phase always returns a :class:PhaseResult — it never re-raises. Callers inspect the result and decide what (if anything) to do with a failed phase.

Functions:

Name Description
run_phase

Run fn under a try/except. NEVER re-raises.
run_phases

Run a list of (name, callable) pairs in order, swallowing per-phase

run_phase 

run_phase(name: str, fn: Callable[..., T], *args: Any, namespace: str = 'phase', alert_severity: str | None = 'error', aggregate_counter: str | None = None, **kwargs: Any) -> PhaseResult[T]

Run fn under a try/except. NEVER re-raises.

Counters bumped (best-effort): - {namespace}.{name}.ok on success - {namespace}.{name}.failed on exception - {namespace}.{name}.{aggregate_counter} += len(result) when aggregate_counter is given AND the result supports len.

Source code in azure_bootstrap/phases/__init__.py 
50
51
52
53
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
def run_phase(
    name: str,
    fn: Callable[..., T],
    *args: Any,
    namespace: str = "phase",
    alert_severity: str | None = "error",
    aggregate_counter: str | None = None,
    **kwargs: Any,
) -> PhaseResult[T]:
    """Run ``fn`` under a try/except. NEVER re-raises.

    Counters bumped (best-effort):
      - ``{namespace}.{name}.ok`` on success
      - ``{namespace}.{name}.failed`` on exception
      - ``{namespace}.{name}.{aggregate_counter} += len(result)`` when
        ``aggregate_counter`` is given AND the result supports ``len``.
    """
    start = time.monotonic()
    try:
        value = fn(*args, **kwargs)
    except BaseException as exc:
        elapsed = round(time.monotonic() - start, 3)
        _logger.exception(
            "Phase %s raised; contributing nothing",
            name,
            extra={
                "phase": name,
                "namespace": namespace,
                "exception_type": type(exc).__name__,
                "elapsed_seconds": elapsed,
            },
        )
        bump_counter(f"{namespace}.{name}.failed")
        if alert_severity:
            _fire_alert(alert_severity, namespace, name, exc)
        return PhaseResult(name=name, ok=False, value=None, exception=exc, elapsed_seconds=elapsed)

    elapsed = round(time.monotonic() - start, 3)
    bump_counter(f"{namespace}.{name}.ok")
    if aggregate_counter is not None:
        try:
            n = len(value)  # type: ignore[arg-type]
            bump_counter(f"{namespace}.{name}.{aggregate_counter}", n)
        except TypeError:
            pass
    return PhaseResult(name=name, ok=True, value=value, elapsed_seconds=elapsed)

run_phases 

run_phases(phases: list[tuple[str, Callable[..., Any]]], *, namespace: str = 'phase', alert_severity: str | None = 'error') -> list[PhaseResult[Any]]

Run a list of (name, callable) pairs in order, swallowing per-phase failures so subsequent phases still execute.

Source code in azure_bootstrap/phases/__init__.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
def run_phases(
    phases: list[tuple[str, Callable[..., Any]]],
    *,
    namespace: str = "phase",
    alert_severity: str | None = "error",
) -> list[PhaseResult[Any]]:
    """Run a list of (name, callable) pairs in order, swallowing per-phase
    failures so subsequent phases still execute."""
    return [
        run_phase(name, fn, namespace=namespace, alert_severity=alert_severity)
        for name, fn in phases
    ]