nat.utils.exception_handlers.automatic_retries#

Attributes#

`T`
`Exc`
`CodePattern`
`logger`
`_CODE_ATTRS`

Functions#

`_extract_status_code`(→ int \| None)	Return a numeric status code found inside exc, else None.
`_pattern_to_regex`(→ re.Pattern[str])	Convert simple wildcard pattern (“4xx”, “5*”, “40x”) to a ^regex$.
`_code_matches`(→ bool)
`_want_retry`(→ bool)	Return True if the exception satisfies either (when provided):
`_retry_decorator`(, retry_codes, retry_on_messages, ...)	Build a decorator that retries with exponential back-off iff:
`patch_with_retry`(, retry_codes, retry_on_messages, ...)	Patch obj instance-locally so every public method retries on failure.

Module Contents#

T#

Exc#

CodePattern#

logger#

_CODE_ATTRS = ('code', 'status', 'status_code', 'http_status')#

_extract_status_code(exc: BaseException) → int | None#: Return a numeric status code found inside exc, else None.

_pattern_to_regex(pat: str) → re.Pattern[str]#: Convert simple wildcard pattern (“4xx”, “5*”, “40x”) to a ^regex$. Rule: ‘x’ or ‘*’ ⇒ any digit.

_code_matches(code: int, pat: CodePattern) → bool#

_want_retry( exc: BaseException, *, code_patterns: collections.abc.Sequence[CodePattern] | None, msg_substrings: collections.abc.Sequence[str] | None, ) → bool#

Return True if the exception satisfies either (when provided):

code_patterns – matches status-code pattern(s)
msg_substrings – contains any of the substrings (case-insensitive)

_retry_decorator( *, retries: int = 3, base_delay: float = 0.25, backoff: float = 2.0, retry_on: Exc = (Exception,), retry_codes: collections.abc.Sequence[CodePattern] | None = None, retry_on_messages: collections.abc.Sequence[str] | None = None, deepcopy: bool = False, ) → collections.abc.Callable[[collections.abc.Callable[Ellipsis, T]], collections.abc.Callable[Ellipsis, T]]#

Build a decorator that retries with exponential back-off iff:

the raised exception is an instance of one of retry_on

AND _want_retry() returns True (i.e. matches codes/messages filters)

If both retry_codes and retry_on_messages are None, all exceptions are retried.

deepcopy:: If True, each retry receives deep‑copied args and **kwargs to avoid mutating shared state between attempts.

patch_with_retry( obj: Any, *, retries: int = 3, base_delay: float = 0.25, backoff: float = 2.0, retry_on: Exc = (Exception,), retry_codes: collections.abc.Sequence[CodePattern] | None = None, retry_on_messages: collections.abc.Sequence[str] | None = None, deepcopy: bool = False, ) → Any#

Patch obj instance-locally so every public method retries on failure.

Extra filters#

retry_codes: Same as before – ints, ranges, or wildcard strings (“4xx”, “5*”…).
retry_on_messages: List of substring patterns. We retry only if any pattern appears (case-insensitive) in str(exc).
deepcopy:: If True, each retry receives deep‑copied args and **kwargs to avoid mutating shared state between attempts.