nat.utils.exception_handlers.automatic_retries#

Attributes#

T

Exc

CodePattern

logger

_CODE_ATTRS

Functions#

_shallow_copy_args(→ tuple[tuple, dict])

Create shallow copies of args and kwargs to avoid deep copy overhead.

_deep_copy_args(→ tuple[tuple, dict])

Create deep copies of args and kwargs to prevent mutation issues.

_clear_exception_context(→ None)

Clear exception traceback to free memory.

_run_gc_if_needed(→ None)

Run garbage collection periodically to free memory.

_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 if:

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#
_shallow_copy_args(args: tuple, kwargs: dict) → tuple[tuple, dict]#

Create shallow copies of args and kwargs to avoid deep copy overhead.

_deep_copy_args(
args: tuple,
kwargs: dict,
skip_first: bool = False,
) → tuple[tuple, dict]#

Create deep copies of args and kwargs to prevent mutation issues.

Args:

args: Positional arguments to copy kwargs: Keyword arguments to copy skip_first: If True, skip copying the first arg (typically ‘self’)

_clear_exception_context(exc: BaseException) → None#

Clear exception traceback to free memory.

_run_gc_if_needed(attempt: int, gc_frequency: int = 3) → None#

Run garbage collection periodically to free memory.

_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,
shallow_copy: bool = True,
gc_frequency: int = 3,
clear_tracebacks: bool = True,
instance_context_aware: bool = False,
) → collections.abc.Callable[[collections.abc.Callable[Ellipsis, T]], collections.abc.Callable[Ellipsis, T]]#

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

  • 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.

instance_context_aware:

If True, the decorator will check for a retry context flag on the first argument (assumed to be ‘self’). If the flag is set, retries are skipped to prevent retry storms in nested method calls.

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,
deep_copy: bool = False,
gc_frequency: int = 3,
clear_tracebacks: bool = True,
) → 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.