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

1"""Resilience pattern protocol class definitions.""" 

2 

3from __future__ import annotations 

4 

5from contextlib import AbstractAsyncContextManager 

6from typing import Any, Protocol, Self, runtime_checkable 

7 

8from lexigram.contracts.infra.resilience.models import ( 

9 CircuitBreakerConfig, 

10 RetryConfig, 

11 TimeoutConfig, 

12) 

13 

14 

15@runtime_checkable 

16class CircuitBreakerProtocol(Protocol): 

17 """Protocol for circuit breaker implementations.""" 

18 

19 @property 

20 def state(self) -> str: 

21 """Get current circuit state (closed, open, half_open).""" 

22 ... 

23 

24 async def call(self, func: Any, *args: Any, **kwargs: Any) -> Any: 

25 """Execute function with circuit breaker protection.""" 

26 ... 

27 

28 def protect(self) -> AbstractAsyncContextManager[None]: 

29 """Return an async context manager that protects a code block. 

30 

31 Raises CircuitOpenError when the circuit is open, and records 

32 success or failure based on the outcome of the protected block. 

33 """ 

34 ... 

35 

36 def reset(self) -> None: 

37 """Reset circuit to closed state.""" 

38 ... 

39 

40 def force_open(self) -> None: 

41 """Force circuit to open state.""" 

42 ... 

43 

44 

45@runtime_checkable 

46class RetryPolicyProtocol(Protocol): 

47 """Protocol for retry policy implementations.""" 

48 

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

50 """Execute function with retry logic.""" 

51 ... 

52 

53 

54@runtime_checkable 

55class BulkheadProtocol(Protocol): 

56 """Protocol for bulkhead implementations.""" 

57 

58 async def __aenter__(self) -> Self: 

59 """Enter bulkhead context.""" 

60 ... 

61 

62 async def __aexit__(self, *args: object) -> None: 

63 """Exit bulkhead context.""" 

64 ... 

65 

66 

67@runtime_checkable 

68class ResiliencePipelineProtocol(Protocol): 

69 """Protocol for resilience pipeline that combines multiple patterns.""" 

70 

71 def add(self, pattern: Any) -> ResiliencePipelineProtocol: 

72 """Add a resilience pattern to the pipeline.""" 

73 ... 

74 

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

76 """Execute function through the resilience pipeline.""" 

77 ... 

78 

79 

80@runtime_checkable 

81class ResiliencePipelineFactoryProtocol(Protocol): 

82 """Factory protocol for creating configured resilience pipelines. 

83 

84 ``lexigram-resilience`` registers a concrete implementation. Other 

85 extension packages (e.g. ``lexigram-sql``) request an optional 

86 ``ResiliencePipelineFactory | None`` via DI injection so they can build 

87 pre-configured pipelines without importing from ``lexigram-resilience`` 

88 directly. 

89 

90 Example:: 

91 

92 class DatabaseResilienceHandler: 

93 def __init__( 

94 self, 

95 pipeline_factory: ResiliencePipelineFactory | None = None, 

96 ) -> None: 

97 self._factory = pipeline_factory 

98 """ 

99 

100 def __call__( 

101 self, 

102 retry_config: RetryConfig, 

103 circuit_config: CircuitBreakerConfig, 

104 timeout_config: TimeoutConfig, 

105 ) -> ResiliencePipelineProtocol: 

106 """Build and return a configured resilience pipeline. 

107 

108 Args: 

109 retry_config: Retry policy settings. 

110 circuit_config: Circuit-breaker settings. 

111 timeout_config: Timeout settings. 

112 

113 Returns: 

114 A configured :class:`ResiliencePipelineProtocol` instance. 

115 """ 

116 ... 

117 

118 

119@runtime_checkable 

120class CircuitBreakerRegistryProtocol(Protocol): 

121 """Protocol for circuit breaker registries.""" 

122 

123 def get(self, name: str) -> CircuitBreakerProtocol | None: 

124 """Get circuit breaker by name.""" 

125 ... 

126 

127 async def get_or_create( 

128 self, 

129 name: str, 

130 config: CircuitBreakerConfig | None = None, 

131 ) -> CircuitBreakerProtocol: 

132 """Get or create circuit breaker by name.""" 

133 ... 

134 

135 def list_breakers(self) -> dict[str, dict[str, Any]]: 

136 """List all circuit breakers.""" 

137 ... 

138 

139 

140@runtime_checkable 

141class ThrottlerProtocol(Protocol): 

142 """Protocol for throttler implementations.""" 

143 

144 async def acquire(self) -> None: 

145 """Acquire permission to proceed.""" 

146 ... 

147 

148 async def try_acquire(self) -> bool: 

149 """Try to acquire permission. Returns True if successful.""" 

150 ... 

151 

152 def get_stats(self) -> dict[str, Any]: 

153 """Get throttling statistics.""" 

154 ... 

155 

156 

157@runtime_checkable 

158class RateLimiterProtocol(Protocol): 

159 """Protocol for token-bucket and sliding-window rate limiters. 

160 

161 ``lexigram-resilience`` is the canonical owner of all rate-limiter 

162 implementations. Extension packages declare this protocol as a 

163 constructor parameter type and receive a concrete implementation via 

164 the DI container. 

165 """ 

166 

167 async def acquire(self) -> None: 

168 """Block until one token is available and consume it.""" 

169 ... 

170 

171 async def try_acquire(self) -> bool: 

172 """Consume one token without blocking. 

173 

174 Returns: 

175 True if the token was acquired; False if the limit is exhausted. 

176 """ 

177 ... 

178 

179 def get_stats(self) -> dict[str, Any]: 

180 """Return current limiter statistics as a plain mapping.""" 

181 ... 

182 

183 

184@runtime_checkable 

185class ResilienceFallbackProtocol(Protocol): 

186 """Executes a sequence of fallback strategies until one succeeds. 

187 

188 Strategies are tried in registration order. The chain raises the 

189 last encountered exception only if all strategies are exhausted. 

190 """ 

191 

192 def add(self, strategy: Any) -> Self: 

193 """Append a fallback strategy to the chain. 

194 

195 Args: 

196 strategy: Callable or coroutine-returning callable to try. 

197 

198 Returns: 

199 Self, for fluent chaining. 

200 """ 

201 ... 

202 

203 async def execute(self) -> Any: 

204 """Execute strategies in order, returning the first success. 

205 

206 Raises: 

207 The last exception if every strategy fails. 

208 """ 

209 ... 

210 

211 

212@runtime_checkable 

213class TimeoutProtocol(Protocol): 

214 """Protocol for timeout policy enforcement. 

215 

216 Implementations enforce maximum execution time budgets on operations. 

217 """ 

218 

219 async def execute_with_timeout( 

220 self, 

221 coro: Any, 

222 timeout_seconds: float, 

223 ) -> Any: 

224 """Execute a coroutine with a timeout budget. 

225 

226 Args: 

227 coro: The coroutine to execute. 

228 timeout_seconds: Maximum allowed seconds before raising. 

229 

230 Raises: 

231 TimeoutError: If execution exceeds the budget. 

232 """ 

233 ... 

234 

235 @property 

236 def default_timeout(self) -> float: 

237 """The default timeout in seconds used when none is specified.""" 

238 ... 

239 

240 

241__all__ = [ 

242 "BulkheadProtocol", 

243 "CircuitBreakerProtocol", 

244 "CircuitBreakerRegistryProtocol", 

245 "RateLimiterProtocol", 

246 "ResilienceFallbackProtocol", 

247 "ResiliencePipelineFactoryProtocol", 

248 "ResiliencePipelineProtocol", 

249 "RetryPolicyProtocol", 

250 "ThrottlerProtocol", 

251 "TimeoutProtocol", 

252]