Coverage for dataexcept/failure_metadata.py: 89%

1"""Machine-readable failure classification for DataExcept exceptions.""" 

2 

3from __future__ import annotations 

4 

5import math 

6from dataclasses import dataclass 

7from typing import Literal, TypeAlias 

8 

9__all__ = ["FailureKind", "FailureMetadata"] 

10 

11FailureKind: TypeAlias = Literal["transient", "permanent", "unknown"] 

12_VALID_FAILURE_KINDS = {"transient", "permanent", "unknown"} 

13 

14 

15@dataclass(frozen=True) 

16class FailureMetadata: 

17 """Describe recovery-relevant properties of an operational failure. 

18 

19 ``failure_kind`` describes whether the underlying condition is known to be 

20 transient, permanent for the same operation/payload, or unknown. 

21 ``retryable`` is deliberately independent: DataExcept describes the 

22 failure, while the calling application still owns retry policy. 

23 """ 

24 

25 failure_kind: FailureKind = "unknown" 

26 retryable: bool | None = None 

27 retry_after_seconds: float | None = None 

28 

29 def __post_init__(self) -> None: 

30 if self.failure_kind not in _VALID_FAILURE_KINDS: 30 ↛ 31line 30 didn't jump to line 31 because the condition on line 30 was never true

31 raise ValueError( 

32 "failure_kind must be 'transient', 'permanent', or 'unknown'" 

33 ) 

34 if self.retryable is not None and not isinstance(self.retryable, bool): 34 ↛ 35line 34 didn't jump to line 35 because the condition on line 34 was never true

35 raise TypeError("retryable must be bool or None") 

36 if self.retry_after_seconds is None: 

37 return 

38 if isinstance(self.retry_after_seconds, bool) or not isinstance( 

39 self.retry_after_seconds, (int, float) 

40 ): 

41 raise TypeError("retry_after_seconds must be a number or None") 

42 seconds = float(self.retry_after_seconds) 

43 if not math.isfinite(seconds) or seconds < 0: 

44 raise ValueError("retry_after_seconds must be finite and non-negative") 

45 object.__setattr__(self, "retry_after_seconds", seconds)