Coverage for src/lexigram/admin/integrations/resilience.py: 78%

1"""Resilience integration — wraps data-source calls with retry/circuit-breaker.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from typing import TYPE_CHECKING, Any 

7 

8if TYPE_CHECKING: 

9 from lexigram.contracts.core.di import ( 

10 ContainerRegistrarProtocol, 

11 ContainerResolverProtocol, 

12 ) 

13 

14 

15@dataclass(frozen=True) 

16class ResilientSpec: 

17 """Specification for resilient data-source calls. 

18 

19 Attributes: 

20 max_retries: Max retry attempts on transient failure. 

21 circuit_breaker: Whether to enable circuit breaker. 

22 timeout_seconds: Per-call timeout. 

23 """ 

24 

25 max_retries: int = 3 

26 circuit_breaker: bool = True 

27 timeout_seconds: float = 30.0 

28 

29 

30class _NoOpResilience: 

31 async def execute(self, func: Any, *args: Any, **kwargs: Any) -> Any: 

32 return await func(*args, **kwargs) 

33 

34 

35class ResilienceIntegration: 

36 """Adapter that decorates data-source calls with retry and circuit-breaker. 

37 

38 Gracefully no-ops when ``lexigram-resilience`` is not installed or the 

39 integration is disabled. 

40 """ 

41 

42 def __init__(self, config: Any) -> None: 

43 self._config = config 

44 self._pipeline: Any = None 

45 self._enabled = False 

46 

47 def register(self, container: ContainerRegistrarProtocol) -> None: 

48 from lexigram.admin.config import ResilienceIntegrationConfig 

49 from lexigram.admin.integrations._optional import is_installed 

50 

51 cfg = self._config 

52 if not isinstance(cfg, ResilienceIntegrationConfig): 

53 cfg = ResilienceIntegrationConfig() 

54 if not cfg.enabled: 

55 self._pipeline = _NoOpResilience() 

56 return 

57 if not is_installed("lexigram.resilience"): 

58 self._pipeline = _NoOpResilience() 

59 return 

60 self._enabled = True 

61 

62 async def boot(self, container: ContainerResolverProtocol) -> None: 

63 if not self._enabled: 

64 return 

65 try: 

66 from lexigram.contracts.infra.resilience import ( 

67 ResiliencePipelineFactoryProtocol, 

68 ) 

69 

70 self._pipeline = await container.resolve(ResiliencePipelineFactoryProtocol) 

71 except Exception: # noqa: BLE001 

72 self._pipeline = _NoOpResilience() 

73 

74 async def shutdown(self) -> None: 

75 pass 

76 

77 async def health_check(self) -> dict[str, Any]: 

78 return { 

79 "status": "healthy" 

80 if not isinstance(self._pipeline, _NoOpResilience) 

81 else "noop" 

82 } 

83 

84 async def execute(self, func: Any, *args: Any, **kwargs: Any) -> Any: 

85 if hasattr(self._pipeline, "execute"): 

86 return await self._pipeline.execute(func, *args, **kwargs) 

87 return await func(*args, **kwargs) 

88 

89 

90__all__ = ["ResilienceIntegration", "ResilientSpec"]