Coverage for src / lexigram / contracts / exceptions / resilience.py: 0%

1"""Resilience pattern exception classes.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from lexigram.contracts.exceptions.base import LexigramError 

8 

9 

10class ResilienceError(LexigramError): 

11 """Base resilience error.""" 

12 

13 _code = "LEX_ERR_RES_001" 

14 

15 def __init__(self, message: str = "Resilience error", **kwargs: Any) -> None: 

16 super().__init__(message, **kwargs) 

17 

18 

19class RetryError(ResilienceError): 

20 """Retry operation error.""" 

21 

22 _code = "LEX_ERR_RES_002" 

23 

24 def __init__(self, message: str = "Retry error", **kwargs: Any) -> None: 

25 super().__init__(message, **kwargs) 

26 

27 

28class CircuitBreakerError(ResilienceError): 

29 """Circuit breaker error.""" 

30 

31 _code = "LEX_ERR_RES_003" 

32 

33 def __init__(self, message: str = "Circuit breaker error", **kwargs: Any) -> None: 

34 super().__init__(message, **kwargs) 

35 

36 

37class BulkheadError(ResilienceError): 

38 """Bulkhead/rejection error.""" 

39 

40 _code = "LEX_ERR_RES_004" 

41 

42 def __init__(self, message: str = "Bulkhead rejected", **kwargs: Any) -> None: 

43 super().__init__(message, **kwargs) 

44 

45 

46class FallbackError(ResilienceError): 

47 """Fallback execution error.""" 

48 

49 _code = "LEX_ERR_RES_005" 

50 

51 def __init__(self, message: str = "Fallback error", **kwargs: Any) -> None: 

52 super().__init__(message, **kwargs) 

53 

54 

55class RetryExhaustedError(RetryError): 

56 """All retry attempts have been exhausted.""" 

57 

58 _code = "LEX_ERR_RES_006" 

59 

60 def __init__( 

61 self, 

62 message: str = "All retry attempts exhausted", 

63 attempts: int = 0, 

64 **kwargs: Any, 

65 ) -> None: 

66 details = kwargs.get("details", {}) 

67 details["attempts"] = attempts 

68 kwargs["details"] = details 

69 super().__init__(message, **kwargs) 

70 self.attempts = attempts 

71 

72 

73class CircuitOpenError(CircuitBreakerError): 

74 """Circuit breaker is open and rejecting requests.""" 

75 

76 _code = "LEX_ERR_RES_007" 

77 

78 def __init__( 

79 self, 

80 message: str = "Circuit breaker is open", 

81 **kwargs: Any, 

82 ) -> None: 

83 super().__init__(message, **kwargs) 

84 

85 

86__all__ = [ 

87 "BulkheadError", 

88 "CircuitBreakerError", 

89 "CircuitOpenError", 

90 "FallbackError", 

91 "ResilienceError", 

92 "RetryError", 

93 "RetryExhaustedError", 

94]