Skip to content

API reference

Models

Typed domain models for SLOs.

These models are the vocabulary of the whole library: an :class:SLO bundles an :class:SLI (how we measure success), a :class:Target (how reliable we promise to be), and a :class:Window (over what rolling period).

Everything downstream — the budget engine, burn-rate math, and alert-rule generation — consumes these models, so they are intentionally small and strict.

AlertingConfig

Bases: BaseModel

Optional per-SLO alerting policy overriding the built-in defaults.

Source code in src/slo_kit/models.py 
151
152
153
154
155
156
class AlertingConfig(BaseModel):
    """Optional per-SLO alerting policy overriding the built-in defaults."""

    model_config = ConfigDict(frozen=True)

    conditions: tuple[BurnCondition, ...] = ()

BurnCondition

Bases: BaseModel

One row of a multi-window multi-burn-rate alert policy.

A condition fires only when the burn rate over both the long and short windows exceeds threshold. The long window gives sensitivity to the right amount of budget burn; the short "for real / still ongoing" window makes the alert resolve quickly once the incident ends.

Source code in src/slo_kit/models.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
class BurnCondition(BaseModel):
    """One row of a multi-window multi-burn-rate alert policy.

    A condition fires only when the burn rate over *both* the long and short
    windows exceeds ``threshold``. The long window gives sensitivity to the
    right amount of budget burn; the short "for real / still ongoing" window
    makes the alert resolve quickly once the incident ends.
    """

    model_config = ConfigDict(frozen=True)

    name: str
    long_window: Window
    short_window: Window
    threshold: float = Field(gt=0.0, description="Burn-rate multiplier, e.g. 14.4.")
    severity: str = Field(default="page")

    @model_validator(mode="after")
    def _validate_windows(self) -> BurnCondition:
        if self.short_window.seconds >= self.long_window.seconds:
            raise ValueError(
                f"{self.name}: short_window ({self.short_window}) must be shorter "
                f"than long_window ({self.long_window})"
            )
        return self

SLI

Bases: BaseModel

A Service-Level Indicator expressed as a good / total event ratio.

Both queries are source-native expressions (PromQL for the Prometheus source). They may contain a {window} placeholder which the source substitutes when it evaluates the SLI over a specific window::

good_query: 'sum(rate(http_requests_total{code!~"5.."}[{window}]))'
total_query: 'sum(rate(http_requests_total[{window}]))'
Source code in src/slo_kit/models.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
class SLI(BaseModel):
    """A Service-Level Indicator expressed as a ``good / total`` event ratio.

    Both queries are source-native expressions (PromQL for the Prometheus
    source). They may contain a ``{window}`` placeholder which the source
    substitutes when it evaluates the SLI over a specific window::

        good_query: 'sum(rate(http_requests_total{code!~"5.."}[{window}]))'
        total_query: 'sum(rate(http_requests_total[{window}]))'
    """

    model_config = ConfigDict(frozen=True)

    good_query: str = Field(
        description="Query returning the count/rate of good (successful) events."
    )
    total_query: str = Field(description="Query returning the count/rate of total (valid) events.")
    description: str = ""

    @model_validator(mode="after")
    def _validate_queries(self) -> SLI:
        if not self.good_query.strip():
            raise ValueError("good_query must not be empty")
        if not self.total_query.strip():
            raise ValueError("total_query must not be empty")
        return self

SLO

Bases: BaseModel

A Service-Level Objective: an SLI held to a Target over a Window.

Source code in src/slo_kit/models.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
class SLO(BaseModel):
    """A Service-Level Objective: an SLI held to a Target over a Window."""

    model_config = ConfigDict(frozen=True)

    name: str = Field(pattern=r"^[a-zA-Z_][a-zA-Z0-9_.-]*$")
    description: str = ""
    service: str = ""
    sli: SLI
    target: Target
    window: Window
    labels: dict[str, str] = Field(default_factory=dict)
    alerting: AlertingConfig | None = None

    @property
    def objective(self) -> float:
        return self.target.objective

    @property
    def error_budget(self) -> float:
        return self.target.error_budget

    def sample(self, source: MetricSource, window: Window | None = None) -> SLISample:
        """Sample good/total event counts for this SLI over ``window``.

        The ``{window}`` placeholder in the SLI queries is substituted with the
        given window (defaulting to the SLO's own window) before querying.
        """
        from .sources.base import SLISample

        win = window or self.window
        good = source.scalar(self.sli.good_query.replace("{window}", win.duration))
        total = source.scalar(self.sli.total_query.replace("{window}", win.duration))
        return SLISample(good=good, total=total, window=win.duration)

    def evaluate(self, source: MetricSource, window: Window | None = None) -> float:
        """Evaluate the SLI over ``window`` and return the compliance ratio."""
        return self.sample(source, window).ratio

evaluate(source, window=None)

Evaluate the SLI over window and return the compliance ratio.

Source code in src/slo_kit/models.py 
194
195
196
def evaluate(self, source: MetricSource, window: Window | None = None) -> float:
    """Evaluate the SLI over ``window`` and return the compliance ratio."""
    return self.sample(source, window).ratio

sample(source, window=None)

Sample good/total event counts for this SLI over window.

The {window} placeholder in the SLI queries is substituted with the given window (defaulting to the SLO's own window) before querying.

Source code in src/slo_kit/models.py 
181
182
183
184
185
186
187
188
189
190
191
192
def sample(self, source: MetricSource, window: Window | None = None) -> SLISample:
    """Sample good/total event counts for this SLI over ``window``.

    The ``{window}`` placeholder in the SLI queries is substituted with the
    given window (defaulting to the SLO's own window) before querying.
    """
    from .sources.base import SLISample

    win = window or self.window
    good = source.scalar(self.sli.good_query.replace("{window}", win.duration))
    total = source.scalar(self.sli.total_query.replace("{window}", win.duration))
    return SLISample(good=good, total=total, window=win.duration)

Target

Bases: BaseModel

The reliability objective, e.g. 0.999 (three nines).

Source code in src/slo_kit/models.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
class Target(BaseModel):
    """The reliability objective, e.g. ``0.999`` (three nines)."""

    model_config = ConfigDict(frozen=True)

    objective: float = Field(
        gt=0.0,
        lt=1.0,
        description="Fraction of good events required, strictly between 0 and 1.",
    )

    @property
    def error_budget(self) -> float:
        """The fraction of events allowed to fail: ``1 - objective``."""
        return 1.0 - self.objective

    def __str__(self) -> str:
        return f"{self.objective:.5g}"

error_budget property

The fraction of events allowed to fail: 1 - objective.

Window

Bases: BaseModel

A rolling time window, e.g. the 30-day SLO window or a 5m alert window.

Source code in src/slo_kit/models.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
class Window(BaseModel):
    """A rolling time window, e.g. the 30-day SLO window or a 5m alert window."""

    model_config = ConfigDict(frozen=True)

    duration: str = Field(description="Prometheus-style duration, e.g. '30d'.")

    @model_validator(mode="after")
    def _validate_duration(self) -> Window:
        parse_duration(self.duration)  # raises on invalid input
        return self

    @property
    def seconds(self) -> float:
        return parse_duration(self.duration)

    @property
    def timedelta(self) -> timedelta:
        return timedelta(seconds=self.seconds)

    def __str__(self) -> str:
        return self.duration

parse_duration(value)

Parse a Prometheus-style duration string into seconds.

parse_duration("5m") 300.0 parse_duration("30d") 2592000.0

Source code in src/slo_kit/models.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
def parse_duration(value: str) -> float:
    """Parse a Prometheus-style duration string into seconds.

    >>> parse_duration("5m")
    300.0
    >>> parse_duration("30d")
    2592000.0
    """
    match = _DURATION_RE.match(value)
    if not match:
        raise ValueError(f"invalid duration {value!r}; expected e.g. '5m', '1h', '30d', '2w'")
    magnitude, unit = match.groups()
    return float(magnitude) * _UNIT_SECONDS[unit]

Spec loading

Load and validate SLO specs from YAML (or plain Python dicts).

The YAML schema mirrors the models but uses friendly scalar fields so specs read naturally::

apiVersion: slo-kit/v1
name: checkout-availability
service: checkout
description: Checkout API availability
objective: 0.999
window: 30d
sli:
  good_query: 'sum(rate(http_requests_total{job="checkout",code!~"5.."}[{window}]))'
  total_query: 'sum(rate(http_requests_total{job="checkout"}[{window}]))'
labels:
  team: payments
alerting:
  conditions:
    - name: fast_burn
      long_window: 1h
      short_window: 5m
      threshold: 14.4
      severity: page

SpecError

Bases: ValueError

Raised when a spec is structurally invalid.

Source code in src/slo_kit/spec.py 
41
42
class SpecError(ValueError):
    """Raised when a spec is structurally invalid."""

dump_spec(slo)

Serialize an :class:SLO back to canonical YAML.

Source code in src/slo_kit/spec.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def dump_spec(slo: SLO) -> str:
    """Serialize an :class:`SLO` back to canonical YAML."""
    doc: dict[str, Any] = {
        "apiVersion": "slo-kit/v1",
        "name": slo.name,
        "service": slo.service,
        "description": slo.description,
        "objective": slo.objective,
        "window": slo.window.duration,
        "sli": {
            "good_query": slo.sli.good_query,
            "total_query": slo.sli.total_query,
        },
    }
    if slo.labels:
        doc["labels"] = dict(slo.labels)
    if slo.alerting and slo.alerting.conditions:
        doc["alerting"] = {
            "conditions": [
                {
                    "name": c.name,
                    "long_window": c.long_window.duration,
                    "short_window": c.short_window.duration,
                    "threshold": c.threshold,
                    "severity": c.severity,
                }
                for c in slo.alerting.conditions
            ]
        }
    return yaml.safe_dump(doc, sort_keys=False)

load_spec(path)

Load a single SLO spec from a YAML file.

Source code in src/slo_kit/spec.py 
107
108
109
110
111
112
113
114
115
116
117
118
def load_spec(path: str | Path) -> SLO:
    """Load a single SLO spec from a YAML file."""
    path = Path(path)
    try:
        raw = yaml.safe_load(path.read_text())
    except FileNotFoundError as exc:
        raise SpecError(f"spec file not found: {path}") from exc
    except yaml.YAMLError as exc:
        raise SpecError(f"could not parse YAML in {path}: {exc}") from exc
    if raw is None:
        raise SpecError(f"spec file is empty: {path}")
    return load_spec_dict(raw)

load_spec_dict(data)

Build a validated :class:SLO from an already-parsed mapping.

Source code in src/slo_kit/spec.py 
 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
100
101
102
103
104
def load_spec_dict(data: dict[str, Any]) -> SLO:
    """Build a validated :class:`SLO` from an already-parsed mapping."""
    if not isinstance(data, dict):
        raise SpecError(f"spec must be a mapping, got {type(data).__name__}")

    api_version = data.get("apiVersion", "slo-kit/v1")
    if api_version not in SUPPORTED_API_VERSIONS:
        raise SpecError(
            f"unsupported apiVersion {api_version!r}; supported: {sorted(SUPPORTED_API_VERSIONS)}"
        )

    sli_raw = data.get("sli")
    if not isinstance(sli_raw, dict):
        raise SpecError("spec is missing an 'sli' mapping")

    if "objective" not in data:
        raise SpecError("spec is missing 'objective'")
    if "window" not in data:
        raise SpecError("spec is missing 'window'")

    alerting = None
    alerting_raw = data.get("alerting")
    if alerting_raw is not None:
        if not isinstance(alerting_raw, dict):
            raise SpecError("'alerting' must be a mapping")
        conditions = tuple(_build_condition(c) for c in alerting_raw.get("conditions", []))
        alerting = AlertingConfig(conditions=conditions)

    try:
        return SLO(
            name=data["name"],
            description=data.get("description", ""),
            service=data.get("service", ""),
            sli=SLI(**sli_raw),
            target=Target(objective=float(data["objective"])),
            window=Window(**_coerce_window(data["window"])),
            labels=data.get("labels", {}) or {},
            alerting=alerting,
        )
    except KeyError as exc:
        raise SpecError(f"spec is missing required field: {exc}") from exc
    except ValidationError as exc:
        raise SpecError(f"invalid spec:\n{exc}") from exc

load_specs(path)

Load one or more SLO specs from a YAML file (supports multi-doc YAML).

Source code in src/slo_kit/spec.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
def load_specs(path: str | Path) -> list[SLO]:
    """Load one or more SLO specs from a YAML file (supports multi-doc YAML)."""
    path = Path(path)
    try:
        docs = list(yaml.safe_load_all(path.read_text()))
    except FileNotFoundError as exc:
        raise SpecError(f"spec file not found: {path}") from exc
    except yaml.YAMLError as exc:
        raise SpecError(f"could not parse YAML in {path}: {exc}") from exc
    specs = [load_spec_dict(doc) for doc in docs if doc is not None]
    if not specs:
        raise SpecError(f"no SLO specs found in {path}")
    return specs

Metric sources

The MetricSource protocol shared by all backends.

A metric source knows one thing: given a backend-native query, return a single scalar value. Everything else — SLI evaluation, budgets, burn rates — is built on top of that primitive, so adding a backend (Prometheus, OTel, a fake for tests) means implementing a single method.

MetricSource

Bases: Protocol

A backend capable of resolving a query to a single float.

The query may contain a {window} placeholder; the caller substitutes a concrete duration before calling :meth:scalar.

Source code in src/slo_kit/sources/base.py 
37
38
39
40
41
42
43
44
45
46
47
@runtime_checkable
class MetricSource(Protocol):
    """A backend capable of resolving a query to a single float.

    The query may contain a ``{window}`` placeholder; the caller substitutes a
    concrete duration before calling :meth:`scalar`.
    """

    def scalar(self, query: str) -> float:
        """Evaluate ``query`` and return its scalar result."""
        ...

scalar(query)

Evaluate query and return its scalar result.

Source code in src/slo_kit/sources/base.py 
45
46
47
def scalar(self, query: str) -> float:
    """Evaluate ``query`` and return its scalar result."""
    ...

SLISample dataclass

Good/total counts for an SLI over some window.

Source code in src/slo_kit/sources/base.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@dataclass(frozen=True)
class SLISample:
    """Good/total counts for an SLI over some window."""

    good: float
    total: float
    window: str

    @property
    def ratio(self) -> float:
        """The good ratio (SLI value); 1.0 when there is no traffic."""
        if self.total <= 0:
            return 1.0
        return min(max(self.good, 0.0), self.total) / self.total

    @property
    def error_ratio(self) -> float:
        return 1.0 - self.ratio

ratio property

The good ratio (SLI value); 1.0 when there is no traffic.

Prometheus metric source.

Talks to the Prometheus HTTP API (/api/v1/query) and reduces the result to a single scalar. Instant vectors with a single sample and scalar results are both supported; multi-sample vectors are summed so that a bare selector like sum(rate(...)) and an un-aggregated one behave sensibly.

PrometheusError

Bases: RuntimeError

Raised when Prometheus returns an error or an unusable response.

Source code in src/slo_kit/sources/prometheus.py 
18
19
class PrometheusError(RuntimeError):
    """Raised when Prometheus returns an error or an unusable response."""

PrometheusSource

A :class:~slo_kit.sources.base.MetricSource backed by Prometheus.

Source code in src/slo_kit/sources/prometheus.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class PrometheusSource:
    """A :class:`~slo_kit.sources.base.MetricSource` backed by Prometheus."""

    def __init__(
        self,
        base_url: str,
        *,
        timeout: float = 30.0,
        client: httpx.Client | None = None,
        headers: dict[str, str] | None = None,
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self._timeout = timeout
        self._owns_client = client is None
        self._client = client or httpx.Client(timeout=timeout, headers=headers or {})

    def scalar(self, query: str) -> float:
        """Run an instant query and collapse the result to a single float."""
        url = f"{self.base_url}/api/v1/query"
        try:
            resp = self._client.get(url, params={"query": query})
            resp.raise_for_status()
        except httpx.HTTPError as exc:
            raise PrometheusError(f"Prometheus request failed: {exc}") from exc

        payload: dict[str, Any] = resp.json()
        if payload.get("status") != "success":
            raise PrometheusError(
                f"Prometheus query failed: {payload.get('error', 'unknown error')}"
            )
        return _reduce_result(payload.get("data", {}))

    def close(self) -> None:
        if self._owns_client:
            self._client.close()

    def __enter__(self) -> PrometheusSource:
        return self

    def __exit__(self, *exc: object) -> None:
        self.close()

scalar(query)

Run an instant query and collapse the result to a single float.

Source code in src/slo_kit/sources/prometheus.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def scalar(self, query: str) -> float:
    """Run an instant query and collapse the result to a single float."""
    url = f"{self.base_url}/api/v1/query"
    try:
        resp = self._client.get(url, params={"query": query})
        resp.raise_for_status()
    except httpx.HTTPError as exc:
        raise PrometheusError(f"Prometheus request failed: {exc}") from exc

    payload: dict[str, Any] = resp.json()
    if payload.get("status") != "success":
        raise PrometheusError(
            f"Prometheus query failed: {payload.get('error', 'unknown error')}"
        )
    return _reduce_result(payload.get("data", {}))

OpenTelemetry metric source.

Most OTel metrics pipelines land in a backend that speaks PromQL (the OTLP metrics data model maps cleanly onto Prometheus, and the Collector's prometheus / prometheusremotewrite exporters are the common path). So the OTel source is a thin specialization of the Prometheus source that applies OTLP naming conventions to the query, keeping the same MetricSource protocol and identical evaluation behaviour.

If your OTel metrics are queried through a backend with a different API, this class is the seam to override :meth:scalar for that API.

OTelSource

Bases: PrometheusSource

OTLP-metrics source backed by a PromQL-compatible query endpoint.

Parameters:

Name Type Description Default
base_url str

PromQL-compatible query endpoint (e.g. a Prometheus, Mimir, or Thanos frontend that ingests OTLP metrics).

 required
normalize_names bool

When true, apply the OTLP -> Prometheus name translation (. -> _) to metric-name-like tokens in the query, matching the Collector's default normalization.

 True
Source code in src/slo_kit/sources/otel.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
class OTelSource(PrometheusSource):
    """OTLP-metrics source backed by a PromQL-compatible query endpoint.

    Args:
        base_url: PromQL-compatible query endpoint (e.g. a Prometheus, Mimir,
            or Thanos frontend that ingests OTLP metrics).
        normalize_names: When true, apply the OTLP -> Prometheus name
            translation (``.`` -> ``_``) to metric-name-like tokens in the
            query, matching the Collector's default normalization.
    """

    def __init__(
        self,
        base_url: str,
        *,
        normalize_names: bool = True,
        timeout: float = 30.0,
        client: httpx.Client | None = None,
        headers: dict[str, str] | None = None,
    ) -> None:
        super().__init__(base_url, timeout=timeout, client=client, headers=headers)
        self._normalize_names = normalize_names

    def scalar(self, query: str) -> float:
        return super().scalar(self._normalize(query))

    def _normalize(self, query: str) -> str:
        """Apply OTLP -> Prometheus metric-name normalization.

        OTLP metric names use dotted namespaces (``http.server.duration``)
        which the Prometheus exporter rewrites to underscores. We only rewrite
        dotted identifiers, leaving float literals (``0.999``) and label
        matchers untouched.
        """
        if not self._normalize_names:
            return query
        import re

        # Dotted identifier: a letter/underscore start, then name.parts, with
        # at least one dot between identifier chars. Avoids matching numbers.
        pattern = re.compile(r"\b([A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)+)\b")
        return pattern.sub(lambda m: m.group(1).replace(".", "_"), query)

Error budget

Error-budget engine.

Given the good/total event counts observed over the SLO window and the objective, this computes how much of the error budget has been consumed, how much remains, and — combined with a current burn rate — an estimated time to exhaustion.

All quantities are derived from three numbers (good, total, objective) so the results are trivially checkable by hand, which is exactly how the tests validate them.

ErrorBudget dataclass

A point-in-time view of an SLO's error budget over its window.

Attributes:

Name Type Description
objective float

The SLO objective, e.g. 0.999.
window_seconds float

Length of the SLO window in seconds.
good float

Observed good events over the window.
total float

Observed total (valid) events over the window.
Source code in src/slo_kit/budget/error_budget.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 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
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
@dataclass(frozen=True)
class ErrorBudget:
    """A point-in-time view of an SLO's error budget over its window.

    Attributes:
        objective: The SLO objective, e.g. ``0.999``.
        window_seconds: Length of the SLO window in seconds.
        good: Observed good events over the window.
        total: Observed total (valid) events over the window.
    """

    objective: float
    window_seconds: float
    good: float
    total: float

    @property
    def error_budget_fraction(self) -> float:
        """The allowed error fraction: ``1 - objective``."""
        return 1.0 - self.objective

    @property
    def allowed_bad_events(self) -> float:
        """Absolute budget: how many events are allowed to fail over the window."""
        return self.error_budget_fraction * self.total

    @property
    def bad_events(self) -> float:
        """Observed bad events (clamped to ``[0, total]``)."""
        good = min(max(self.good, 0.0), self.total)
        return self.total - good

    @property
    def consumed(self) -> float:
        """Absolute number of budgeted failures already spent."""
        return self.bad_events

    @property
    def remaining(self) -> float:
        """Absolute budget remaining (never negative)."""
        return max(0.0, self.allowed_bad_events - self.consumed)

    @property
    def consumed_pct(self) -> float:
        """Fraction of the budget consumed in ``[0, 1+]``.

        Values above 1.0 mean the budget is overspent (SLO violated). If the
        budget is zero (objective of 1.0 is disallowed, but total==0 yields a
        zero budget), any failure counts as fully consumed.
        """
        allowed = self.allowed_bad_events
        if allowed <= 0:
            return 0.0 if self.bad_events == 0 else float("inf")
        return self.consumed / allowed

    @property
    def remaining_pct(self) -> float:
        """Fraction of budget remaining in ``[0, 1]`` (clamped)."""
        consumed = self.consumed_pct
        if consumed == float("inf"):
            return 0.0
        return max(0.0, 1.0 - consumed)

    @property
    def is_exhausted(self) -> bool:
        """True once consumed failures meet or exceed the allowed budget."""
        return self.consumed >= self.allowed_bad_events and self.bad_events > 0

    @property
    def sli(self) -> float:
        """The observed SLI (good ratio) over the window."""
        if self.total <= 0:
            return 1.0
        return 1.0 - error_rate_from_counts(self.good, self.total)

    @property
    def current_burn_rate(self) -> float:
        """Burn rate implied by the full-window error rate."""
        return burn_rate_from_counts(self.good, self.total, self.objective)

    def time_to_exhaustion(self, burn_rate: float | None = None) -> timedelta | None:
        """Estimate time until the *remaining* budget is exhausted.

        At a constant burn rate ``B`` the entire budget empties in
        ``window / B``; the remaining fraction therefore empties in
        ``remaining_pct * window / B``.

        Args:
            burn_rate: Burn rate to project forward. Defaults to the
                full-window :attr:`current_burn_rate`.

        Returns:
            A ``timedelta``, or ``None`` if the budget is already exhausted or
            the burn rate is non-positive (never exhausts).
        """
        rate = self.current_burn_rate if burn_rate is None else burn_rate
        if rate <= 0:
            return None
        if self.remaining <= 0:
            return timedelta(0)
        seconds = self.remaining_pct * self.window_seconds / rate
        return timedelta(seconds=seconds)

allowed_bad_events property

Absolute budget: how many events are allowed to fail over the window.

bad_events property

Observed bad events (clamped to [0, total]).

consumed property

Absolute number of budgeted failures already spent.

consumed_pct property

Fraction of the budget consumed in [0, 1+].

Values above 1.0 mean the budget is overspent (SLO violated). If the budget is zero (objective of 1.0 is disallowed, but total==0 yields a zero budget), any failure counts as fully consumed.

current_burn_rate property

Burn rate implied by the full-window error rate.

error_budget_fraction property

The allowed error fraction: 1 - objective.

is_exhausted property

True once consumed failures meet or exceed the allowed budget.

remaining property

Absolute budget remaining (never negative).

remaining_pct property

Fraction of budget remaining in [0, 1] (clamped).

sli property

The observed SLI (good ratio) over the window.

time_to_exhaustion(burn_rate=None)

Estimate time until the remaining budget is exhausted.

At a constant burn rate B the entire budget empties in window / B; the remaining fraction therefore empties in remaining_pct * window / B.

Parameters:

Name Type Description Default
burn_rate float | None

Burn rate to project forward. Defaults to the full-window :attr:current_burn_rate.

 None

Returns:

Type Description
timedelta | None

A timedelta, or None if the budget is already exhausted or
timedelta | None

the burn rate is non-positive (never exhausts).
Source code in src/slo_kit/budget/error_budget.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def time_to_exhaustion(self, burn_rate: float | None = None) -> timedelta | None:
    """Estimate time until the *remaining* budget is exhausted.

    At a constant burn rate ``B`` the entire budget empties in
    ``window / B``; the remaining fraction therefore empties in
    ``remaining_pct * window / B``.

    Args:
        burn_rate: Burn rate to project forward. Defaults to the
            full-window :attr:`current_burn_rate`.

    Returns:
        A ``timedelta``, or ``None`` if the budget is already exhausted or
        the burn rate is non-positive (never exhausts).
    """
    rate = self.current_burn_rate if burn_rate is None else burn_rate
    if rate <= 0:
        return None
    if self.remaining <= 0:
        return timedelta(0)
    seconds = self.remaining_pct * self.window_seconds / rate
    return timedelta(seconds=seconds)

compute_error_budget(*, objective, window_seconds, good, total)

Construct an :class:ErrorBudget (keyword-only for call-site clarity).

Source code in src/slo_kit/budget/error_budget.py 
127
128
129
130
131
132
133
134
135
136
137
def compute_error_budget(
    *, objective: float, window_seconds: float, good: float, total: float
) -> ErrorBudget:
    """Construct an :class:`ErrorBudget` (keyword-only for call-site clarity)."""
    if not 0.0 < objective < 1.0:
        raise ValueError(f"objective must be in (0, 1), got {objective}")
    if window_seconds <= 0:
        raise ValueError("window_seconds must be positive")
    if total < 0 or good < 0:
        raise ValueError("good and total counts must be non-negative")
    return ErrorBudget(objective=objective, window_seconds=window_seconds, good=good, total=total)

Burn rate

Burn-rate math — the beating heart of SLO alerting.

Definition. The burn rate is how fast we are consuming the error budget relative to the pace that would exactly exhaust it over the SLO window:

burn_rate = observed_error_rate / error_budget
          = observed_error_rate / (1 - objective)

where observed_error_rate is the fraction of bad events measured over some (usually short) window.

Why divide by the error budget? The error budget is the maximum error rate you can sustain for the entire SLO window and still exactly meet the objective. So:

* burn_rate == 1  -> you are on track to spend 100% of the budget over the
                     whole window (right at the objective).
* burn_rate == 2  -> you would exhaust the whole budget in half the window.
* burn_rate == 14.4 over 30 days -> the whole budget gone in ~50 hours,
                     which is why 14.4 is the classic fast-page threshold.

The general identity: at a constant burn rate B, the full budget is consumed in window / B. Every threshold in the multi-window tables is just a choice of "how much of the budget are we willing to let a single incident burn before we page?" — see :mod:slo_kit.alerts.multiwindow.

This module is pure arithmetic on counts/ratios: no I/O, fully deterministic, and exhaustively unit-tested against hand-computed values.

budget_consumed_fraction(threshold, alert_window_seconds, slo_window_seconds)

Fraction of the total budget consumed if a burn of threshold runs for exactly alert_window_seconds.

This is the "budget consumed before alert" column in the SRE workbook tables: threshold * alert_window / slo_window.

Source code in src/slo_kit/budget/burn_rate.py 
85
86
87
88
89
90
91
92
93
94
95
96
def budget_consumed_fraction(
    threshold: float, alert_window_seconds: float, slo_window_seconds: float
) -> float:
    """Fraction of the total budget consumed if a burn of ``threshold`` runs
    for exactly ``alert_window_seconds``.

    This is the "budget consumed before alert" column in the SRE workbook
    tables: ``threshold * alert_window / slo_window``.
    """
    if slo_window_seconds <= 0:
        raise ValueError("slo_window_seconds must be positive")
    return threshold * alert_window_seconds / slo_window_seconds

burn_rate(error_rate, objective)

Burn rate from an already-computed error rate and the objective.

burn_rate = error_rate / (1 - objective).

Source code in src/slo_kit/budget/burn_rate.py 
56
57
58
59
60
61
62
63
64
def burn_rate(error_rate: float, objective: float) -> float:
    """Burn rate from an already-computed error rate and the objective.

    ``burn_rate = error_rate / (1 - objective)``.
    """
    if not 0.0 < objective < 1.0:
        raise ValueError(f"objective must be in (0, 1), got {objective}")
    budget = 1.0 - objective
    return error_rate / budget

burn_rate_from_counts(good, total, objective)

Burn rate computed directly from good/total event counts.

Source code in src/slo_kit/budget/burn_rate.py 
67
68
69
def burn_rate_from_counts(good: float, total: float, objective: float) -> float:
    """Burn rate computed directly from good/total event counts."""
    return burn_rate(error_rate_from_counts(good, total), objective)

error_rate_from_counts(good, total)

Fraction of bad events: (total - good) / total.

A total of zero means "no traffic", which we treat as a 0.0 error rate (you cannot fail requests you never received). good is clamped to [0, total] so noisy inputs cannot produce a negative or >1 error rate.

Source code in src/slo_kit/budget/burn_rate.py 
43
44
45
46
47
48
49
50
51
52
53
def error_rate_from_counts(good: float, total: float) -> float:
    """Fraction of bad events: ``(total - good) / total``.

    A total of zero means "no traffic", which we treat as a 0.0 error rate
    (you cannot fail requests you never received). ``good`` is clamped to
    ``[0, total]`` so noisy inputs cannot produce a negative or >1 error rate.
    """
    if total <= 0:
        return 0.0
    good = min(max(good, 0.0), total)
    return (total - good) / total

threshold_for_budget_fraction(budget_fraction, alert_window_seconds, slo_window_seconds)

Inverse of :func:budget_consumed_fraction.

Given the fraction of budget you're willing to burn within an alert window, return the burn-rate threshold that corresponds to it. Handy for deriving custom multi-window tables for non-standard SLO windows.

Source code in src/slo_kit/budget/burn_rate.py 
 99
100
101
102
103
104
105
106
107
108
109
110
def threshold_for_budget_fraction(
    budget_fraction: float, alert_window_seconds: float, slo_window_seconds: float
) -> float:
    """Inverse of :func:`budget_consumed_fraction`.

    Given the fraction of budget you're willing to burn within an alert window,
    return the burn-rate threshold that corresponds to it. Handy for deriving
    custom multi-window tables for non-standard SLO windows.
    """
    if alert_window_seconds <= 0:
        raise ValueError("alert_window_seconds must be positive")
    return budget_fraction * slo_window_seconds / alert_window_seconds

time_to_full_burn_seconds(current_burn_rate, window_seconds)

Seconds to exhaust a full, fresh budget at current_burn_rate.

Returns None when the burn rate is non-positive (budget never exhausts). Note this is time to burn the entire budget; to account for a budget that is already partly consumed, use :meth:slo_kit.budget.error_budget.ErrorBudget.time_to_exhaustion.

Source code in src/slo_kit/budget/burn_rate.py 
72
73
74
75
76
77
78
79
80
81
82
def time_to_full_burn_seconds(current_burn_rate: float, window_seconds: float) -> float | None:
    """Seconds to exhaust a *full, fresh* budget at ``current_burn_rate``.

    Returns ``None`` when the burn rate is non-positive (budget never
    exhausts). Note this is time to burn the *entire* budget; to account for a
    budget that is already partly consumed, use
    :meth:`slo_kit.budget.error_budget.ErrorBudget.time_to_exhaustion`.
    """
    if current_burn_rate <= 0:
        return None
    return window_seconds / current_burn_rate

Multi-window alerting

Multi-window, multi-burn-rate alerting — the differentiating core.

This implements the alerting strategy from the Google SRE Workbook ("Alerting on SLOs", multiwindow multi-burn-rate). The problem it solves:

  • Alert on a single long window (e.g. burn rate over 1h) and you detect slow burns but page slowly and keep paging long after an incident ends (poor reset time).
  • Alert on a single short window and you page fast but with lots of false positives from brief blips.

The fix is to require two conditions simultaneously: a long window (the one that defines how much budget is being burned) AND a short window (a fraction of the long one — typically 1/12) that must also be burning. The short window is the "is this still happening right now?" gate: it makes the alert fire quickly, and — crucially — stop firing quickly once errors subside, because the short window recovers long before the long window does.

We then stack several such (long, short, threshold) conditions at different severities so a catastrophic burn pages immediately while a slow leak opens a ticket:

| severity | long | short | burn rate | budget burned before firing |
|----------|------|-------|-----------|-----------------------------|
| page     | 1h   | 5m    | 14.4      | 2%   (over a 30d window)    |
| page     | 6h   | 30m   | 6         | 5%                          |
| ticket   | 24h  | 2h    | 3         | 10%                         |

A :class:MultiWindowPolicy fires when any of its conditions fire, and each condition fires only when the burn rate over both its windows exceeds its threshold.

The module is pure logic over a mapping of window -> burn_rate; it does no I/O and is validated against an explicit truth table in the tests.

ConditionResult dataclass

Outcome of evaluating a single :class:BurnCondition.

Source code in src/slo_kit/alerts/multiwindow.py 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@dataclass(frozen=True)
class ConditionResult:
    """Outcome of evaluating a single :class:`BurnCondition`."""

    condition: BurnCondition
    long_burn_rate: float
    short_burn_rate: float
    firing: bool

    @property
    def name(self) -> str:
        return self.condition.name

    @property
    def severity(self) -> str:
        return self.condition.severity

MultiWindowPolicy

A set of multi-window burn-rate conditions evaluated together.

Source code in src/slo_kit/alerts/multiwindow.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
class MultiWindowPolicy:
    """A set of multi-window burn-rate conditions evaluated together."""

    def __init__(self, conditions: Iterable[BurnCondition] | None = None) -> None:
        conds = tuple(conditions) if conditions is not None else default_conditions()
        if not conds:
            raise ValueError("a MultiWindowPolicy needs at least one condition")
        self.conditions: tuple[BurnCondition, ...] = conds

    @property
    def required_windows(self) -> tuple[str, ...]:
        """All distinct window durations this policy needs burn rates for."""
        seen: dict[str, None] = {}
        for c in self.conditions:
            seen.setdefault(c.long_window.duration, None)
            seen.setdefault(c.short_window.duration, None)
        return tuple(seen)

    def evaluate(self, burn_rates: Mapping[str, float]) -> PolicyResult:
        """Evaluate the policy against a mapping of ``window -> burn_rate``.

        Each condition fires iff the burn rate over *both* its long and short
        windows is strictly greater than the condition's threshold.
        """
        results = []
        for cond in self.conditions:
            long_key = cond.long_window.duration
            short_key = cond.short_window.duration
            if long_key not in burn_rates:
                raise KeyError(f"missing burn rate for window {long_key!r}")
            if short_key not in burn_rates:
                raise KeyError(f"missing burn rate for window {short_key!r}")
            long_br = burn_rates[long_key]
            short_br = burn_rates[short_key]
            firing = long_br > cond.threshold and short_br > cond.threshold
            results.append(
                ConditionResult(
                    condition=cond,
                    long_burn_rate=long_br,
                    short_burn_rate=short_br,
                    firing=firing,
                )
            )
        return PolicyResult(results=tuple(results))

    def is_firing(self, burn_rates: Mapping[str, float]) -> bool:
        return self.evaluate(burn_rates).firing

required_windows property

All distinct window durations this policy needs burn rates for.

evaluate(burn_rates)

Evaluate the policy against a mapping of window -> burn_rate.

Each condition fires iff the burn rate over both its long and short windows is strictly greater than the condition's threshold.

Source code in src/slo_kit/alerts/multiwindow.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def evaluate(self, burn_rates: Mapping[str, float]) -> PolicyResult:
    """Evaluate the policy against a mapping of ``window -> burn_rate``.

    Each condition fires iff the burn rate over *both* its long and short
    windows is strictly greater than the condition's threshold.
    """
    results = []
    for cond in self.conditions:
        long_key = cond.long_window.duration
        short_key = cond.short_window.duration
        if long_key not in burn_rates:
            raise KeyError(f"missing burn rate for window {long_key!r}")
        if short_key not in burn_rates:
            raise KeyError(f"missing burn rate for window {short_key!r}")
        long_br = burn_rates[long_key]
        short_br = burn_rates[short_key]
        firing = long_br > cond.threshold and short_br > cond.threshold
        results.append(
            ConditionResult(
                condition=cond,
                long_burn_rate=long_br,
                short_burn_rate=short_br,
                firing=firing,
            )
        )
    return PolicyResult(results=tuple(results))

PolicyResult dataclass

Outcome of evaluating a whole :class:MultiWindowPolicy.

Source code in src/slo_kit/alerts/multiwindow.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@dataclass(frozen=True)
class PolicyResult:
    """Outcome of evaluating a whole :class:`MultiWindowPolicy`."""

    results: tuple[ConditionResult, ...]

    @property
    def firing(self) -> bool:
        return any(r.firing for r in self.results)

    @property
    def firing_conditions(self) -> tuple[ConditionResult, ...]:
        return tuple(r for r in self.results if r.firing)

    @property
    def severity(self) -> str | None:
        """Highest-priority severity currently firing, or ``None``.

        ``page`` outranks ``ticket``; unknown severities sort last.
        """
        firing = self.firing_conditions
        if not firing:
            return None
        order = {"page": 0, "ticket": 1}
        return min((r.severity for r in firing), key=lambda s: order.get(s, 99))

severity property

Highest-priority severity currently firing, or None.

page outranks ticket; unknown severities sort last.

default_conditions()

The canonical SRE-workbook conditions for a 30-day SLO window.

Two paging conditions (fast + medium burn) and one ticket condition (slow burn). Callers can override these per-SLO via AlertingConfig.

Source code in src/slo_kit/alerts/multiwindow.py 
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
def default_conditions() -> tuple[BurnCondition, ...]:
    """The canonical SRE-workbook conditions for a 30-day SLO window.

    Two paging conditions (fast + medium burn) and one ticket condition
    (slow burn). Callers can override these per-SLO via ``AlertingConfig``.
    """
    return (
        BurnCondition(
            name="fast_burn",
            long_window=Window(duration="1h"),
            short_window=Window(duration="5m"),
            threshold=14.4,
            severity="page",
        ),
        BurnCondition(
            name="slow_burn",
            long_window=Window(duration="6h"),
            short_window=Window(duration="30m"),
            threshold=6.0,
            severity="page",
        ),
        BurnCondition(
            name="slower_burn",
            long_window=Window(duration="24h"),
            short_window=Window(duration="2h"),
            threshold=3.0,
            severity="ticket",
        ),
    )

Prometheus rule generation

Generate Prometheus recording + alerting rules from an SLO.

Given an SLO and its multi-window policy we emit a Prometheus rule group with:

  • one recording rule per distinct window, recording the SLI error ratio over that window as slo:sli_error:ratio_rate<window>. Recording rules keep the alert expressions cheap and readable.
  • one alerting rule per burn condition. Because burn_rate = error_ratio / (1 - objective), the condition burn_rate > threshold is equivalent to error_ratio > threshold * (1 - objective) — so the alert compares the recorded error ratios against threshold * error_budget directly, with no division in the alert expression.

The output is a plain dict that serializes to the exact YAML Prometheus expects under groups:. Golden-file tests pin the rendered output.

build_rule_group(slo)

Build the Prometheus rule group (as a dict) for a single SLO.

Source code in src/slo_kit/alerts/prometheus_rules.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def build_rule_group(slo: SLO) -> dict[str, Any]:
    """Build the Prometheus rule group (as a dict) for a single SLO."""
    policy = _resolve_policy(slo)
    labels = _base_labels(slo)
    budget = slo.error_budget

    rules: list[dict[str, Any]] = []

    for window in _distinct_windows(policy):
        rules.append(
            {
                "record": recording_rule_name(window),
                "expr": _render_error_ratio_expr(slo, window),
                "labels": {**labels, "window": window},
            }
        )

    for cond in policy.conditions:
        long_metric = recording_rule_name(cond.long_window.duration)
        short_metric = recording_rule_name(cond.short_window.duration)
        # burn_rate > threshold  <=>  error_ratio > threshold * (1 - objective)
        limit = cond.threshold * budget
        label_selector = f'{{slo="{slo.name}"}}'
        expr = (
            f"(\n"
            f"  {long_metric}{label_selector} > {limit:g}\n"
            f"  and\n"
            f"  {short_metric}{label_selector} > {limit:g}\n"
            f")"
        )
        rules.append(
            {
                "alert": _alert_name(slo, cond),
                "expr": expr,
                "labels": {**labels, "severity": cond.severity, "condition": cond.name},
                "annotations": {
                    "summary": (
                        f"SLO {slo.name}: {cond.name} burn rate exceeded ({cond.threshold:g}x)"
                    ),
                    "description": (
                        f"Error budget for SLO '{slo.name}' is burning at more than "
                        f"{cond.threshold:g}x over {cond.long_window.duration} and "
                        f"{cond.short_window.duration}. Objective: {slo.objective:g}."
                    ),
                },
            }
        )

    return {"name": f"slo:{slo.name}", "rules": rules}

generate_rules(slo)

Return the full {'groups': [...]} structure for one SLO.

Source code in src/slo_kit/alerts/prometheus_rules.py 
128
129
130
def generate_rules(slo: SLO) -> dict[str, Any]:
    """Return the full ``{'groups': [...]}`` structure for one SLO."""
    return {"groups": [build_rule_group(slo)]}

generate_rules_yaml(slo)

Render the Prometheus rules for an SLO to a YAML string.

Source code in src/slo_kit/alerts/prometheus_rules.py 
133
134
135
def generate_rules_yaml(slo: SLO) -> str:
    """Render the Prometheus rules for an SLO to a YAML string."""
    return yaml.safe_dump(generate_rules(slo), sort_keys=False, default_flow_style=False)

recording_rule_name(window_duration)

Metric name of the recorded error ratio for a window, e.g. ...rate5m.

Source code in src/slo_kit/alerts/prometheus_rules.py 
37
38
39
def recording_rule_name(window_duration: str) -> str:
    """Metric name of the recorded error ratio for a window, e.g. ``...rate5m``."""
    return f"{_ERROR_RATIO_PREFIX}{window_duration}"

Status

Programmatic SLO status: budget + burn rates + alert policy, one call.

:func:evaluate_status samples an SLO against a metric source and returns a :class:SLOStatus bundling everything a human or dashboard needs: the current SLI, the error-budget breakdown, per-window burn rates, projected time to exhaustion, and whether the multi-window alert policy is firing. It also serializes to JSON for the CLI and Grafana.

SLOStatus dataclass

A complete point-in-time status report for one SLO.

Source code in src/slo_kit/report/status.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 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
 96
 97
 98
 99
100
101
102
103
104
105
@dataclass(frozen=True)
class SLOStatus:
    """A complete point-in-time status report for one SLO."""

    slo: SLO
    budget: ErrorBudget
    burn_rates: dict[str, float] = field(default_factory=dict)
    policy_result: PolicyResult | None = None

    @property
    def sli(self) -> float:
        return self.budget.sli

    @property
    def objective(self) -> float:
        return self.slo.objective

    @property
    def budget_remaining_pct(self) -> float:
        return self.budget.remaining_pct

    @property
    def budget_consumed_pct(self) -> float:
        return self.budget.consumed_pct

    @property
    def is_exhausted(self) -> bool:
        return self.budget.is_exhausted

    @property
    def is_firing(self) -> bool:
        return bool(self.policy_result and self.policy_result.firing)

    @property
    def severity(self) -> str | None:
        return self.policy_result.severity if self.policy_result else None

    def burn_rate(self, window: str) -> float:
        """Burn rate over a given window duration (must have been sampled)."""
        if window not in self.burn_rates:
            raise KeyError(
                f"no burn rate sampled for window {window!r}; available: {sorted(self.burn_rates)}"
            )
        return self.burn_rates[window]

    def to_dict(self) -> dict[str, Any]:
        """JSON-serializable status structure."""
        ttl = self.budget.time_to_exhaustion()
        return {
            "slo": self.slo.name,
            "service": self.slo.service,
            "objective": self.objective,
            "window": self.slo.window.duration,
            "sli": self.sli,
            "error_budget": {
                "allowed_bad_events": self.budget.allowed_bad_events,
                "consumed": self.budget.consumed,
                "remaining": self.budget.remaining,
                "consumed_pct": self.budget.consumed_pct,
                "remaining_pct": self.budget.remaining_pct,
                "is_exhausted": self.budget.is_exhausted,
                "time_to_exhaustion_seconds": ttl.total_seconds() if ttl else None,
            },
            "burn_rates": dict(self.burn_rates),
            "alerts": {
                "firing": self.is_firing,
                "severity": self.severity,
                "conditions": [
                    {
                        "name": r.name,
                        "severity": r.severity,
                        "firing": r.firing,
                        "long_window": r.condition.long_window.duration,
                        "short_window": r.condition.short_window.duration,
                        "threshold": r.condition.threshold,
                        "long_burn_rate": r.long_burn_rate,
                        "short_burn_rate": r.short_burn_rate,
                    }
                    for r in (self.policy_result.results if self.policy_result else ())
                ],
            },
        }

burn_rate(window)

Burn rate over a given window duration (must have been sampled).

Source code in src/slo_kit/report/status.py 
61
62
63
64
65
66
67
def burn_rate(self, window: str) -> float:
    """Burn rate over a given window duration (must have been sampled)."""
    if window not in self.burn_rates:
        raise KeyError(
            f"no burn rate sampled for window {window!r}; available: {sorted(self.burn_rates)}"
        )
    return self.burn_rates[window]

to_dict()

JSON-serializable status structure.

Source code in src/slo_kit/report/status.py 
 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
100
101
102
103
104
105
def to_dict(self) -> dict[str, Any]:
    """JSON-serializable status structure."""
    ttl = self.budget.time_to_exhaustion()
    return {
        "slo": self.slo.name,
        "service": self.slo.service,
        "objective": self.objective,
        "window": self.slo.window.duration,
        "sli": self.sli,
        "error_budget": {
            "allowed_bad_events": self.budget.allowed_bad_events,
            "consumed": self.budget.consumed,
            "remaining": self.budget.remaining,
            "consumed_pct": self.budget.consumed_pct,
            "remaining_pct": self.budget.remaining_pct,
            "is_exhausted": self.budget.is_exhausted,
            "time_to_exhaustion_seconds": ttl.total_seconds() if ttl else None,
        },
        "burn_rates": dict(self.burn_rates),
        "alerts": {
            "firing": self.is_firing,
            "severity": self.severity,
            "conditions": [
                {
                    "name": r.name,
                    "severity": r.severity,
                    "firing": r.firing,
                    "long_window": r.condition.long_window.duration,
                    "short_window": r.condition.short_window.duration,
                    "threshold": r.condition.threshold,
                    "long_burn_rate": r.long_burn_rate,
                    "short_burn_rate": r.short_burn_rate,
                }
                for r in (self.policy_result.results if self.policy_result else ())
            ],
        },
    }

evaluate_status(slo, source)

Sample slo against source and build a full :class:SLOStatus.

Source code in src/slo_kit/report/status.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def evaluate_status(slo: SLO, source: MetricSource) -> SLOStatus:
    """Sample ``slo`` against ``source`` and build a full :class:`SLOStatus`."""
    # 1. Error budget over the full SLO window.
    window_sample = slo.sample(source)
    budget = compute_error_budget(
        objective=slo.objective,
        window_seconds=slo.window.seconds,
        good=window_sample.good,
        total=window_sample.total,
    )

    # 2. Burn rates over each window the alert policy needs.
    policy = _resolve_policy(slo)
    burn_rates: dict[str, float] = {}
    for duration in policy.required_windows:
        sample = slo.sample(source, Window(duration=duration))
        burn_rates[duration] = burn_rate_from_counts(sample.good, sample.total, slo.objective)

    policy_result = policy.evaluate(burn_rates)
    return SLOStatus(slo=slo, budget=budget, burn_rates=burn_rates, policy_result=policy_result)

Deploy gate

CI deploy gate: block deploys when the error budget is spent.

The pattern: run slo-kit gate in a deploy pipeline. If the SLO's error budget is exhausted (or has fallen below a configured floor, or an alert is already firing), the gate exits non-zero and the deploy is blocked — you don't ship risky changes while you're already burning reliability you don't have.

GateDecision dataclass

The outcome of a deploy-gate evaluation.

Source code in src/slo_kit/gate.py 
20
21
22
23
24
25
26
27
28
29
30
31
@dataclass(frozen=True)
class GateDecision:
    """The outcome of a deploy-gate evaluation."""

    passed: bool
    reason: str
    status: SLOStatus

    @property
    def exit_code(self) -> int:
        """Process exit code: 0 = allow deploy, 1 = block."""
        return 0 if self.passed else 1

exit_code property

Process exit code: 0 = allow deploy, 1 = block.

evaluate_gate(slo, source, *, min_budget_pct=0.0, block_on_firing=False)

Decide whether a deploy should proceed for slo.

Parameters:

Name Type Description Default
slo SLO

The SLO to gate on.

 required
source MetricSource

Metric source to sample.

 required
min_budget_pct float

Minimum remaining budget fraction (0..1) required to pass. 0.0 blocks only on full exhaustion; 0.2 requires at least 20% of the budget remaining.

 0.0
block_on_firing bool

When true, also block if the multi-window alert policy is currently firing, even if budget remains.

 False
Source code in src/slo_kit/gate.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
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
def evaluate_gate(
    slo: SLO,
    source: MetricSource,
    *,
    min_budget_pct: float = 0.0,
    block_on_firing: bool = False,
) -> GateDecision:
    """Decide whether a deploy should proceed for ``slo``.

    Args:
        slo: The SLO to gate on.
        source: Metric source to sample.
        min_budget_pct: Minimum remaining budget fraction (``0..1``) required
            to pass. ``0.0`` blocks only on full exhaustion; ``0.2`` requires
            at least 20% of the budget remaining.
        block_on_firing: When true, also block if the multi-window alert
            policy is currently firing, even if budget remains.
    """
    status = evaluate_status(slo, source)
    remaining = status.budget_remaining_pct

    if remaining <= min_budget_pct:
        return GateDecision(
            passed=False,
            reason=(
                f"error budget too low: {remaining:.1%} remaining (minimum {min_budget_pct:.1%})"
            ),
            status=status,
        )

    if block_on_firing and status.is_firing:
        return GateDecision(
            passed=False,
            reason=f"burn-rate alert firing (severity: {status.severity})",
            status=status,
        )

    return GateDecision(
        passed=True,
        reason=f"error budget healthy: {remaining:.1%} remaining",
        status=status,
    )