Edit on GitHub

agent_search_gateway.retry

Generic configurable exponential retry engine.

 1"""Generic configurable exponential retry engine."""
 2
 3import asyncio
 4from collections.abc import Awaitable, Callable
 5from contextlib import suppress
 6from typing import TypeVar
 7
 8from .models import RetryPolicy
 9
10T = TypeVar("T")
11BeforeAttempt = Callable[[int], None]
12OnRetry = Callable[[int, BaseException, float], None]
13
14
15async def retry_async(
16    policy: RetryPolicy,
17    operation: Callable[[], Awaitable[T]],
18    *,
19    is_retryable: Callable[[BaseException], bool],
20    sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
21    before_attempt: BeforeAttempt | None = None,
22    on_retry: OnRetry | None = None,
23) -> T:
24    for attempt in range(1, policy.max_attempts + 1):
25        if before_attempt is not None:
26            with suppress(Exception):
27                before_attempt(attempt)
28        try:
29            return await operation()
30        except BaseException as exc:
31            if isinstance(exc, asyncio.CancelledError):
32                raise
33            if attempt == policy.max_attempts or not is_retryable(exc):
34                raise
35            delay = min(
36                policy.base_delay_seconds * (2 ** (attempt - 1)),
37                policy.max_delay_seconds,
38            )
39            if on_retry is not None:
40                with suppress(Exception):
41                    on_retry(attempt, exc, delay)
42            await sleep(delay)
43    raise RuntimeError("retry loop exhausted unexpectedly")
BeforeAttempt = collections.abc.Callable[[int], None]
OnRetry = collections.abc.Callable[[int, BaseException, float], None]
async def retry_async( policy: agent_search_gateway.models.RetryPolicy, operation: Callable[[], Awaitable[~T]], *, is_retryable: Callable[[BaseException], bool], sleep: Callable[[float], Awaitable[None]] = <function sleep>, before_attempt: Callable[[int], None] | None = None, on_retry: Callable[[int, BaseException, float], None] | None = None) -> ~T: 
16async def retry_async(
17    policy: RetryPolicy,
18    operation: Callable[[], Awaitable[T]],
19    *,
20    is_retryable: Callable[[BaseException], bool],
21    sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
22    before_attempt: BeforeAttempt | None = None,
23    on_retry: OnRetry | None = None,
24) -> T:
25    for attempt in range(1, policy.max_attempts + 1):
26        if before_attempt is not None:
27            with suppress(Exception):
28                before_attempt(attempt)
29        try:
30            return await operation()
31        except BaseException as exc:
32            if isinstance(exc, asyncio.CancelledError):
33                raise
34            if attempt == policy.max_attempts or not is_retryable(exc):
35                raise
36            delay = min(
37                policy.base_delay_seconds * (2 ** (attempt - 1)),
38                policy.max_delay_seconds,
39            )
40            if on_retry is not None:
41                with suppress(Exception):
42                    on_retry(attempt, exc, delay)
43            await sleep(delay)
44    raise RuntimeError("retry loop exhausted unexpectedly")