Python API

Section 8 · From docs/RESILIENCE.md on GitHub

Everything below is exported from cloudg.resilience. cloudg.retry re-exports call_with_resilience, call_with_resilience_sync, RetryPolicy, Scope, classify, retry_after and ErrorKind for callers of the old module.

8.1 call_with_resilience

await call_with_resilience(fn, *args, scope, policy=None, governor=None,
                           sleep=None, clock=None, **kwargs)
call_with_resilience_sync(fn, *args, scope, policy=None, governor=None,
                          sleep=time.sleep, clock=None, **kwargs)

The async version awaits fn if it is a coroutine function and runs anything else in a worker thread with asyncio.to_thread. The sync version is for code that already runs in a worker thread. Both apply the full sequence from section 2 and re-raise the original exception when they give up. When the provider is disabled, fn is called directly with no retries.

import asyncio

from cloudg.resilience import Scope, call_with_resilience, get_governor, stats_scope


class TooManyRequests(Exception):
    """Stands in for any SDK error: classify() reads status_code and headers."""

    status_code = 429

    def __init__(self, retry_after: str) -> None:
        super().__init__("429 Too Many Requests")
        self.headers = {"Retry-After": retry_after}


attempts = 0


async def list_widgets(page: int) -> list[str]:
    global attempts
    attempts += 1
    if attempts < 3:
        raise TooManyRequests("1")
    return [f"widget-{page}-{i}" for i in range(3)]


async def main() -> None:
    scope = Scope("aws", "123456789012", "eu-west-1", "widgets")
    with stats_scope() as stats:
        items = await call_with_resilience(list_widgets, 1, scope=scope)
    print(items, "after", attempts, "attempts")
    print(stats.messages())
    print(get_governor().limiter.rate(scope))


asyncio.run(main())

['widget-1-0', 'widget-1-1', 'widget-1-2'] after 3 attempts
['aws/123456789012/eu-west-1/widgets: throttled 2x, slowed to 5.0 rps, 2 retries']
5.0

RetryPolicy overrides the provider's settings for one call. Fields left at None use the provider's values:

Field	Default	Meaning
max_retries	provider's max_retries	retries after the first attempt
base_delay	0.5	lower bound of the jitter
max_backoff	provider's max_backoff_seconds	cap of one sleep
deadline	provider's deadline_seconds	overall time; 0 disables
use_budget	True	draw retries from the provider's retry budget
retry_transient	True	False retries throttling only

8.2 @resilient

The decorator form. scope is a Scope or a callable that receives the function's arguments and returns one. Coroutine functions get call_with_resilience, plain functions call_with_resilience_sync.

from cloudg.resilience import RetryPolicy, Scope, resilient


class Busy(Exception):
    status_code = 503


calls = {"n": 0}


@resilient(
    lambda project, **_: Scope("gcp", f"projects/{project}", None, "compute"),
    policy=RetryPolicy(max_retries=4, max_backoff=2.0, deadline=30.0),
)
def list_instances(project: str) -> list[str]:
    calls["n"] += 1
    if calls["n"] == 1:
        raise Busy("service overloaded")
    return ["vm-1", "vm-2"]


print(list_instances("demo-project"), calls["n"], "calls")

['vm-1', 'vm-2'] 2 calls

The 503 classified as THROTTLED, so besides the retry the gcp/projects/demo-project/*/compute scope was slowed down for every other caller in the process.

8.3 The governor and run statistics

from cloudg import CloudGConfig
from cloudg.inventory import InventoryMapper
from cloudg.resilience import configure, get_governor, stats_scope

config = CloudGConfig(providers=["aws"])
configure(config)                      # done for you by InventoryMapper / MultiAccountCollector

with stats_scope() as stats:           # also done by InventoryMapper.map_inventory()
    ...                                # any cloudg collection, or your own SDK calls

report = get_governor().summary(stats) # this run: totals, messages, skipped, scopes,
                                       # plus the process's tripped breakers and slowed buckets
lifetime = get_governor().summary()    # everything since the governor was created

result = InventoryMapper(config).map_inventory_sync()
result.throttling                      # the same summary for that run, or None

Lower-level pieces: get_governor().limiter.snapshot() (every bucket's rate, ceiling, tokens, capacity, pause and throttle count), get_governor().breakers.snapshot(only_tripped=False), get_governor().budget("aws").remaining, get_governor().try_retry(scope) (take one retry-budget token, False when it is spent), get_governor().breaker_scope(scope) and get_governor().limiter.leaf_scope(scope) (the scope at the granularity of its bucket and breaker), get_governor().limiter.bulkhead(scope) (a Bulkhead with acquire(), acquire_sync(), try_acquire(), release() and in_use; cloudg.resilience.limiter.Slot wraps one acquired slot that is released exactly once), classify(exc), retry_after(exc), describe_error(exc), is_throttle(exc). Your own boto3 sessions can join the shared limiter with cloudg.resilience.aws.install_aws_hooks(session, account_id, region), your own azure-mgmt clients with cloudg.resilience.azure.build_azure_client(ClientClass, credential, subscription_id, subscription_id=subscription_id).

8.4 LiveOperationGuard

import asyncio

from cloudg.resilience import CooldownActive, LiveOperationGuard, operation_key


async def collect() -> str:
    await asyncio.sleep(0.2)
    return "dataset-1"


async def main() -> None:
    guard = LiveOperationGuard(cooldown_seconds=120, max_concurrent_per_scope=1)
    key = operation_key("map", ["aws"], ["aws/123456789012"])
    scopes = ["aws/123456789012"]

    # Two identical requests at once: one collection, both get its result
    a, b = await asyncio.gather(
        guard.run(key, collect, scopes=scopes, caller="alice"),
        guard.run(key, collect, scopes=scopes, caller="bob"),
    )
    print(a, b, guard.status()["started"], "started,", guard.status()["joined"], "joined")

    try:
        await guard.run(key, collect, scopes=scopes, caller="alice")
    except CooldownActive as exc:
        print(exc.to_dict())

    # An operator override skips the cooldown (but not the concurrency caps)
    print(await guard.run(key, collect, scopes=scopes, caller="ops", bypass_cooldown=True))


asyncio.run(main())

dataset-1 dataset-1 1 started, 1 joined
{'reason': 'cooldown', 'message': 'live collection of aws/123456789012 ran recently; cooling down, retry in 120s or use the cached dataset', 'retry_after_seconds': 120.0, 'scopes': ['aws/123456789012'], 'caller': 'alice'}
dataset-1

run(key, factory, *, scopes=(), caller=None, bypass_cooldown=False, wait=False, timeout=None) takes a zero-argument factory that returns the awaitable, not the awaitable itself, so a refused call never creates a coroutine. Other members: check(scopes, caller=..., bypass_cooldown=...) raises the rejection a new call would get, without side effects; cooldown_remaining(scope); in_flight(key); reset(scope=None) clears one cooldown, or every cooldown and quota; mark_completed(scopes) starts a cooldown after a collection that ran outside the guard. Construct it with clock= to test it without waiting.