Skip to content

ratelimit

azure_bootstrap.ratelimit

In-process token-bucket rate limiter.

Belt-and-suspenders for L7 rate limiting at ingress (Istio EnvoyFilter is fine; this defends against a sidecar-wedged or local-dev scenario where the ingress filter isn't in the path).

The :func:fastapi_rate_limit helper builds a dependency callable that returns 429 with an empty body on rejection — detail strings leak budget state.

Classes:

Name Description
MultiUnitLimiter

Multi-unit sliding window limiter (pages/records/seconds/chars/calls).

Functions:

Name Description
fastapi_rate_limit

FastAPI dependency factory. 429 + empty body on rejection.
webhook_bucket

Preset for Microsoft-Graph-style webhooks: 240 burst, 4/s sustained.
admin_bucket

Preset for manual-trigger endpoints: 30 burst, 0.5/s sustained.

MultiUnitLimiter 

MultiUnitLimiter(*, limits: dict[str, tuple[float, float]], fail_closed: bool | None = None, name: str = 'multi')

Multi-unit sliding window limiter (pages/records/seconds/chars/calls).

Source code in azure_bootstrap/ratelimit/__init__.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def __init__(
    self,
    *,
    limits: dict[str, tuple[float, float]],
    fail_closed: bool | None = None,
    name: str = "multi",
) -> None:
    self._buckets = {
        unit: TokenBucket(budget=budget, refill_per_second=refill, name=f"{name}.{unit}")
        for unit, (budget, refill) in limits.items()
    }
    if fail_closed is None:
        import os

        fail_closed = os.environ.get("RATE_LIMIT_FAIL_CLOSED", "0") == "1"
    self._fail_closed = fail_closed

fastapi_rate_limit 

fastapi_rate_limit(bucket: TokenBucket, *, detail: str | None = None) -> Callable[..., Any]

FastAPI dependency factory. 429 + empty body on rejection.

detail is intentionally ignored by default — leak-resistant. Passing a string opts into showing it (not recommended for public endpoints).

Source code in azure_bootstrap/ratelimit/__init__.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def fastapi_rate_limit(
    bucket: TokenBucket,
    *,
    detail: str | None = None,
) -> Callable[..., Any]:
    """FastAPI dependency factory. 429 + empty body on rejection.

    ``detail`` is intentionally ignored by default — leak-resistant. Passing
    a string opts into showing it (not recommended for public endpoints).
    """

    async def _check() -> None:
        if bucket.consume(1.0):
            return
        from fastapi import HTTPException  # type: ignore[import-not-found]

        raise HTTPException(status_code=429, detail=detail)

    return _check

webhook_bucket 

webhook_bucket(*, name: str = 'webhook') -> TokenBucket

Preset for Microsoft-Graph-style webhooks: 240 burst, 4/s sustained.

Source code in azure_bootstrap/ratelimit/__init__.py 
94
95
96
def webhook_bucket(*, name: str = "webhook") -> TokenBucket:
    """Preset for Microsoft-Graph-style webhooks: 240 burst, 4/s sustained."""
    return TokenBucket(budget=240.0, refill_per_second=4.0, name=name)

admin_bucket 

admin_bucket(*, name: str = 'admin') -> TokenBucket

Preset for manual-trigger endpoints: 30 burst, 0.5/s sustained.

Source code in azure_bootstrap/ratelimit/__init__.py 
 99
100
101
def admin_bucket(*, name: str = "admin") -> TokenBucket:
    """Preset for manual-trigger endpoints: 30 burst, 0.5/s sustained."""
    return TokenBucket(budget=30.0, refill_per_second=0.5, name=name)