Coverage for src/taipanstack/resilience/circuit_breaker.py: 100%

1""" 

2Circuit Breaker pattern implementation. 

3 

4Provides protection against cascading failures by temporarily 

5blocking calls to a failing service. Compatible with any 

6Python framework (sync and async). 

7""" 

8 

9import contextlib 

10import functools 

11import inspect 

12import logging 

13import math 

14import threading 

15import time 

16from collections.abc import Awaitable, Callable 

17from dataclasses import dataclass, field 

18from enum import Enum 

19from typing import ( 

20 TYPE_CHECKING, 

21 ParamSpec, 

22 Protocol, 

23 TypeGuard, 

24 TypeVar, 

25 cast, 

26 overload, 

27) 

28 

29from taipanstack.core.result import Err 

30 

31P = ParamSpec("P") 

32R = TypeVar("R") 

33 

34 

35class CircuitBreakerDecorator(Protocol): 

36 """Protocol for the circuit breaker decorator.""" 

37 

38 @overload 

39 def __call__(self, func: Callable[P, R]) -> Callable[P, R]: ... 

40 

41 @overload 

42 def __call__( 

43 self, 

44 func: Callable[P, Awaitable[R]], 

45 ) -> Callable[P, Awaitable[R]]: ... 

46 

47 

48logger = logging.getLogger("taipanstack.resilience.circuit_breaker") 

49 

50if TYPE_CHECKING: 

51 

52 class StructlogProtocol(Protocol): 

53 """Protocol for structural logging to appease static checkers.""" 

54 

55 def error(self, _event: str, **kwargs: object) -> None: 

56 """Log an error event.""" 

57 ... 

58 

59 def warning(self, _event: str, **kwargs: object) -> None: 

60 """Log a warning event.""" 

61 ... 

62else: 

63 StructlogProtocol = object 

64 

65try: 

66 import structlog as _structlog 

67 

68 _structlog_logger: StructlogProtocol | None = cast( 

69 StructlogProtocol, 

70 _structlog.get_logger("taipanstack.resilience.circuit_breaker"), 

71 ) 

72 _HAS_STRUCTLOG = True 

73except ImportError: 

74 _structlog_logger = None 

75 _HAS_STRUCTLOG = False 

76 

77 

78class CircuitState(Enum): 

79 """States of the circuit breaker.""" 

80 

81 CLOSED = "closed" # Normal operation, requests flow through 

82 OPEN = "open" # Circuit is tripped, requests are blocked 

83 HALF_OPEN = "half_open" # Testing if service has recovered 

84 

85 

86def _safe_str(e: Exception) -> str: 

87 try: 

88 return str(e) 

89 except Exception: 

90 try: 

91 return repr(e) 

92 except Exception: 

93 return "<unprintable exception>" 

94 

95 

96class CircuitBreakerError(Exception): 

97 """Raised when circuit breaker is open.""" 

98 

99 def __init__(self, message: str, state: CircuitState) -> None: 

100 """Initialize CircuitBreakerError. 

101 

102 Args: 

103 message: Error description. 

104 state: Current circuit state. 

105 

106 """ 

107 self.state = state 

108 super().__init__(message) 

109 

110 

111@dataclass(frozen=True) 

112class CircuitBreakerConfig: 

113 """Configuration for circuit breaker behavior. 

114 

115 Attributes: 

116 failure_threshold: Number of failures before opening circuit. 

117 success_threshold: Successes needed in half-open to close. 

118 timeout: Seconds before trying half-open after open. 

119 excluded_exceptions: Exceptions that don't count as failures. 

120 failure_exceptions: Exceptions that count as failures. 

121 

122 """ 

123 

124 failure_threshold: int = 5 

125 success_threshold: int = 2 

126 timeout: float = 30.0 

127 excluded_exceptions: tuple[type[Exception], ...] = () 

128 failure_exceptions: tuple[type[Exception], ...] = (Exception,) 

129 

130 def _check_finite(self, value: object, name: str) -> None: 

131 if type(value) not in (int, float): 

132 raise TypeError(f"{name} must be of type int or float") 

133 # type narrowing for type checker 

134 val = float(value) # type: ignore[arg-type] 

135 if not math.isfinite(val) or val < 0: 

136 raise ValueError(f"{name} must be finite and non-negative") 

137 

138 def __post_init__(self) -> None: 

139 """Validate configuration values.""" 

140 self._check_finite(self.failure_threshold, "failure_threshold") 

141 self._check_finite(self.success_threshold, "success_threshold") 

142 self._check_finite(self.timeout, "timeout") 

143 

144 

145@dataclass 

146class CircuitBreakerState: 

147 """Internal state tracking for circuit breaker.""" 

148 

149 state: CircuitState = CircuitState.CLOSED 

150 failure_count: int = 0 

151 success_count: int = 0 

152 half_open_attempts: int = 0 

153 last_failure_time: float = 0.0 

154 lock: threading.Lock = field(default_factory=threading.Lock) 

155 

156 

157class CircuitBreaker: 

158 """Circuit breaker implementation. 

159 

160 Monitors function calls and opens the circuit when too many 

161 failures occur, preventing further calls until the service 

162 recovers. Supports both sync and async functions. 

163 

164 Example: 

165 >>> breaker = CircuitBreaker(failure_threshold=3) 

166 >>> @breaker 

167 ... def call_external_api(): 

168 ... return requests.get("https://api.example.com", timeout=10) 

169 

170 """ 

171 

172 @staticmethod 

173 def _is_valid_metric(value: object, min_val: float = 0) -> TypeGuard[int | float]: 

174 """Check if metric is a valid finite number >= min_val.""" 

175 if not isinstance(value, (int, float)): 

176 return False 

177 try: 

178 return math.isfinite(value) and value >= min_val 

179 except Exception: 

180 return False 

181 

182 @staticmethod 

183 def _get_safe_threshold(value: object, min_val: float, default: float) -> float: 

184 """Return the threshold if valid, otherwise return the default.""" 

185 if CircuitBreaker._is_valid_metric(value, min_val): 

186 return float(value) 

187 return default 

188 

189 @staticmethod 

190 def _check_finite_val(value: float, min_val: float, err_msg: str) -> None: 

191 if ( 

192 not isinstance(value, (int, float)) 

193 or not math.isfinite(value) 

194 or value < min_val 

195 ): 

196 raise ValueError(err_msg) 

197 

198 @staticmethod 

199 def _validate_thresholds( 

200 timeout: float, 

201 failure_threshold: int, 

202 success_threshold: int, 

203 ) -> None: 

204 CircuitBreaker._check_finite_val( 

205 timeout, 

206 0, 

207 "timeout must be a finite non-negative number", 

208 ) 

209 CircuitBreaker._check_finite_val( 

210 failure_threshold, 

211 1, 

212 "failure_threshold must be a finite number >= 1", 

213 ) 

214 CircuitBreaker._check_finite_val( 

215 success_threshold, 

216 1, 

217 "success_threshold must be a finite number >= 1", 

218 ) 

219 

220 def __init__( 

221 self, 

222 *, 

223 failure_threshold: int = 5, 

224 success_threshold: int = 2, 

225 timeout: float = 30.0, 

226 excluded_exceptions: tuple[type[Exception], ...] = (), 

227 failure_exceptions: tuple[type[Exception], ...] = (Exception,), 

228 name: str = "default", 

229 on_state_change: Callable[[CircuitState, CircuitState], None] | None = None, 

230 ) -> None: 

231 """Initialize CircuitBreaker. 

232 

233 Args: 

234 failure_threshold: Failures before opening circuit. 

235 success_threshold: Successes to close from half-open. 

236 timeout: Seconds before attempting half-open. 

237 excluded_exceptions: Exceptions that don't trip circuit. 

238 failure_exceptions: Exceptions that count as failures. 

239 name: Name for logging/identification. 

240 on_state_change: Optional callback invoked on state transitions 

241 with (old_state, new_state). Useful for custom monitoring. 

242 

243 """ 

244 CircuitBreaker._validate_thresholds( 

245 timeout, 

246 failure_threshold, 

247 success_threshold, 

248 ) 

249 

250 self.config = CircuitBreakerConfig( 

251 failure_threshold=failure_threshold, 

252 success_threshold=success_threshold, 

253 timeout=timeout, 

254 excluded_exceptions=excluded_exceptions, 

255 failure_exceptions=failure_exceptions, 

256 ) 

257 self.name = name 

258 self._state = CircuitBreakerState() 

259 self._on_state_change = on_state_change 

260 

261 @property 

262 def state(self) -> CircuitState: 

263 """Get current circuit state.""" 

264 return self._state.state 

265 

266 @property 

267 def failure_count(self) -> int: 

268 """Get current failure count.""" 

269 return self._state.failure_count 

270 

271 def _log_callback_failure( 

272 self, 

273 old_state: CircuitState, 

274 new_state: CircuitState, 

275 e: Exception, 

276 ) -> None: 

277 if _HAS_STRUCTLOG and _structlog_logger is not None: 

278 _structlog_logger.error( 

279 "circuit_state_change_callback_failed", 

280 circuit=self.name, 

281 old_state=old_state.value, 

282 new_state=new_state.value, 

283 error=_safe_str(e), 

284 ) 

285 else: 

286 logger.error( 

287 "Circuit %s state change callback failed: %s", 

288 self.name, 

289 _safe_str(e), 

290 ) 

291 

292 def _log_structlog_warning( 

293 self, 

294 old_state: CircuitState, 

295 new_state: CircuitState, 

296 ) -> None: 

297 if _HAS_STRUCTLOG and _structlog_logger is not None: 

298 _structlog_logger.warning( 

299 "circuit_state_changed", 

300 circuit=self.name, 

301 old_state=old_state.value, 

302 new_state=new_state.value, 

303 failure_count=self._state.failure_count, 

304 ) 

305 

306 def _notify_state_change( 

307 self, 

308 old_state: CircuitState, 

309 new_state: CircuitState, 

310 ) -> None: 

311 """Notify callback of state transition if registered. 

312 

313 Emit a structured log via structlog when no callback is provided 

314 and structlog is available. 

315 """ 

316 if self._on_state_change is not None: 

317 try: 

318 self._on_state_change(old_state, new_state) 

319 except Exception as e: 

320 self._log_callback_failure(old_state, new_state, e) 

321 return 

322 

323 self._log_structlog_warning(old_state, new_state) 

324 

325 def _get_safe_timeout(self) -> float: 

326 try: 

327 return float(self.config.timeout) 

328 except (TypeError, ValueError): 

329 return 30.0 

330 

331 def _calculate_safe_elapsed( 

332 self, parsed_failure: float | int, safe_timeout: float, now: float 

333 ) -> float: 

334 try: 

335 val = float(parsed_failure) 

336 if not math.isfinite(val): 

337 return safe_timeout 

338 elapsed = now - val 

339 if elapsed < 0: 

340 return safe_timeout 

341 return elapsed 

342 except Exception: 

343 return safe_timeout 

344 

345 def _compute_elapsed( 

346 self, last_failure: float | None, safe_timeout: float, now: float 

347 ) -> float | None: 

348 if not isinstance(last_failure, (int, float)): 

349 return None 

350 return self._calculate_safe_elapsed(last_failure, safe_timeout, now) 

351 

352 def _calculate_elapsed_time(self, now: float) -> float | None: 

353 """Calculate time elapsed since last failure.""" 

354 return self._compute_elapsed( 

355 self._state.last_failure_time, self._get_safe_timeout(), now 

356 ) 

357 

358 def _transition_to_half_open( 

359 self, 

360 elapsed: float, 

361 ) -> tuple[bool, tuple[CircuitState, CircuitState] | None]: 

362 """Transition the circuit to half-open state.""" 

363 # Before transitioning, verify if we can make an attempt 

364 # This happens in a lock, so it's thread-safe. However, once 

365 # the state changes to HALF_OPEN, subsequent threads in the 

366 # same lock block will hit the HALF_OPEN case. 

367 self._state.state = CircuitState.HALF_OPEN 

368 self._state.success_count = 0 

369 # Initialize half_open_attempts to 1 because this first call 

370 # that transitions the state is also an attempt. 

371 self._state.half_open_attempts = 1 

372 logger.info( 

373 "Circuit %s entering half-open state (was open for %.1fs, failures=%d)", 

374 self.name, 

375 elapsed, 

376 self._state.failure_count, 

377 ) 

378 return True, (CircuitState.OPEN, CircuitState.HALF_OPEN) 

379 

380 def _evaluate_open_timeout( 

381 self, 

382 elapsed: float, 

383 ) -> tuple[bool, tuple[CircuitState, CircuitState] | None]: 

384 timeout = self._get_safe_threshold(self.config.timeout, 0, 30.0) 

385 if elapsed >= timeout: 

386 return self._transition_to_half_open(elapsed) 

387 return False, None 

388 

389 def _get_valid_elapsed(self) -> float | None: 

390 try: 

391 now = time.monotonic() 

392 except Exception: 

393 return None 

394 if not isinstance(now, (int, float)) or not math.isfinite(now): 

395 return None 

396 return self._calculate_elapsed_time(now) 

397 

398 def _handle_open_state( 

399 self, 

400 ) -> tuple[bool, tuple[CircuitState, CircuitState] | None]: 

401 """Handle logic for OPEN state in _should_attempt.""" 

402 elapsed = self._get_valid_elapsed() 

403 if elapsed is None: 

404 return False, None 

405 return self._evaluate_open_timeout(elapsed) 

406 

407 def _handle_attempt_half_open(self) -> bool: 

408 if not self._is_valid_metric(self._state.half_open_attempts): 

409 return False 

410 

411 success_threshold = self._get_safe_threshold( 

412 self.config.success_threshold, 

413 1, 

414 2, 

415 ) 

416 

417 if self._state.half_open_attempts < success_threshold: 

418 self._state.half_open_attempts += 1 

419 return True 

420 return False 

421 

422 def _acquire_lock(self) -> bool: 

423 """Safely attempt to acquire the lock.""" 

424 try: 

425 return self._state.lock.acquire(timeout=0.1) 

426 except Exception: 

427 return False 

428 

429 def _safe_release(self) -> None: 

430 with contextlib.suppress(Exception): 

431 self._state.lock.release() 

432 

433 def _evaluate_state_for_attempt( 

434 self, 

435 ) -> tuple[bool, tuple[CircuitState, CircuitState] | None]: 

436 state = self._state.state 

437 if state is CircuitState.CLOSED: 

438 return True, None 

439 if state is CircuitState.OPEN: 

440 return self._handle_open_state() 

441 if state is CircuitState.HALF_OPEN: 

442 return self._handle_attempt_half_open(), None 

443 return False, None # type: ignore[unreachable] 

444 

445 def _should_attempt(self) -> bool: 

446 """Check if a call should be attempted.""" 

447 if not self._acquire_lock(): 

448 return False 

449 try: 

450 should_attempt, state_change = self._evaluate_state_for_attempt() 

451 finally: 

452 self._safe_release() 

453 

454 if state_change: 

455 self._notify_state_change(*state_change) 

456 

457 return should_attempt 

458 

459 def _handle_success_half_open(self) -> tuple[CircuitState, CircuitState] | None: 

460 if not self._is_valid_metric(self._state.success_count): 

461 # Type corruption detected, reset and increment 

462 self._state.success_count = 1 

463 else: 

464 self._state.success_count += 1 

465 

466 success_threshold = self._get_safe_threshold( 

467 self.config.success_threshold, 

468 1, 

469 2, 

470 ) 

471 

472 if self._state.success_count >= success_threshold: 

473 self._state.state = CircuitState.CLOSED 

474 self._state.failure_count = 0 

475 self._state.half_open_attempts = 0 

476 logger.info( 

477 "Circuit %s closed after recovery (%d consecutive successes)", 

478 self.name, 

479 self._state.success_count, 

480 ) 

481 return (CircuitState.HALF_OPEN, CircuitState.CLOSED) 

482 return None 

483 

484 def _get_success_state_change(self) -> tuple[CircuitState, CircuitState] | None: 

485 state = self._state.state 

486 if state is CircuitState.HALF_OPEN: 

487 return self._handle_success_half_open() 

488 if state is CircuitState.CLOSED: 

489 # Reset failure count on success 

490 self._state.failure_count = 0 

491 return None 

492 return None 

493 

494 def _record_success(self) -> None: 

495 """Record a successful call.""" 

496 if not self._acquire_lock(): 

497 return 

498 try: 

499 state_change = self._get_success_state_change() 

500 finally: 

501 self._safe_release() 

502 

503 if state_change: 

504 self._notify_state_change(*state_change) 

505 

506 def _handle_failure_half_open(self) -> tuple[CircuitState, CircuitState] | None: 

507 """Handle failure when in HALF_OPEN state.""" 

508 self._state.state = CircuitState.OPEN 

509 self._state.half_open_attempts = 0 

510 logger.warning( 

511 "Circuit %s reopened after failure in half-open", 

512 self.name, 

513 ) 

514 return (CircuitState.HALF_OPEN, CircuitState.OPEN) 

515 

516 def _handle_failure_closed(self) -> tuple[CircuitState, CircuitState] | None: 

517 """Handle failure when in CLOSED state.""" 

518 # Check against corrupted NaN/Inf failure_count 

519 if not self._is_valid_metric(self._state.failure_count): 

520 self._state.state = CircuitState.OPEN 

521 logger.warning( 

522 "Circuit %s opened due to state/type corruption in failure_count", 

523 self.name, 

524 ) 

525 return (CircuitState.CLOSED, CircuitState.OPEN) 

526 

527 failure_threshold = self._get_safe_threshold( 

528 self.config.failure_threshold, 

529 1, 

530 5, 

531 ) 

532 

533 if self._state.failure_count >= failure_threshold: 

534 self._state.state = CircuitState.OPEN 

535 logger.warning( 

536 "Circuit %s opened after %d failures (threshold=%d)", 

537 self.name, 

538 self._state.failure_count, 

539 failure_threshold, 

540 ) 

541 return (CircuitState.CLOSED, CircuitState.OPEN) 

542 

543 return None 

544 

545 def _increment_failure_count(self) -> None: 

546 if not self._is_valid_metric(self._state.failure_count): 

547 # Handle type mutation (e.g. failure_count became string) or Inf/NaN 

548 # Safe degradation: reset to max so it opens immediately 

549 self._state.failure_count = self.config.failure_threshold 

550 else: 

551 self._state.failure_count += 1 

552 

553 def _update_failure_metrics(self) -> None: 

554 self._increment_failure_count() 

555 

556 try: 

557 now = time.monotonic() 

558 except Exception: 

559 now = float("nan") 

560 

561 if isinstance(now, (int, float)) and math.isfinite(now): 

562 self._state.last_failure_time = now 

563 

564 def _get_failure_state_change(self) -> tuple[CircuitState, CircuitState] | None: 

565 """Get state change for a failure.""" 

566 if self._state.state is CircuitState.HALF_OPEN: 

567 return self._handle_failure_half_open() 

568 if self._state.state is CircuitState.CLOSED: 

569 return self._handle_failure_closed() 

570 return None 

571 

572 def _is_excluded_exception(self, exc: Exception) -> bool: 

573 try: 

574 return isinstance(exc, self.config.excluded_exceptions) 

575 except TypeError: 

576 return False 

577 

578 def _record_failure(self, exc: Exception) -> None: 

579 """Record a failed call.""" 

580 if self._is_excluded_exception(exc): 

581 return 

582 

583 state_change: tuple[CircuitState, CircuitState] | None = None 

584 

585 if not self._acquire_lock(): 

586 return 

587 try: 

588 self._update_failure_metrics() 

589 state_change = self._get_failure_state_change() 

590 finally: 

591 self._safe_release() 

592 

593 if state_change: 

594 self._notify_state_change(*state_change) 

595 

596 def reset(self) -> None: 

597 """Reset circuit breaker to closed state.""" 

598 state_change: tuple[CircuitState, CircuitState] | None = None 

599 if not self._acquire_lock(): 

600 return 

601 try: 

602 old_state = self._state.state 

603 if old_state is not CircuitState.CLOSED: 

604 state_change = (old_state, CircuitState.CLOSED) 

605 

606 self._state.state = CircuitState.CLOSED 

607 self._state.failure_count = 0 

608 self._state.success_count = 0 

609 self._state.half_open_attempts = 0 

610 logger.info("Circuit %s manually reset", self.name) 

611 finally: 

612 self._safe_release() 

613 

614 if state_change: 

615 self._notify_state_change(*state_change) 

616 

617 def _is_failure_exception(self, exc: Exception) -> bool: 

618 try: 

619 return isinstance(exc, self.config.failure_exceptions) 

620 except TypeError: 

621 return True 

622 

623 def _process_result(self, result: R) -> R: 

624 """Process Result outcome and record success/failure. 

625 

626 Args: 

627 result: The result to process. 

628 

629 Returns: 

630 The original result. 

631 

632 """ 

633 if isinstance(result, Err): 

634 err_val = result.unwrap_err() 

635 if self._is_failure_exception(err_val): 

636 self._record_failure(err_val) 

637 return result 

638 # Ignored exception in Result monad 

639 return result 

640 self._record_success() 

641 return result 

642 

643 def _is_valid_half_open_attempts(self) -> bool: 

644 """Validate half-open attempts value against corruption.""" 

645 val = self._state.half_open_attempts 

646 if not isinstance(val, (int, float)): 

647 return False # type: ignore[unreachable] 

648 try: 

649 return math.isfinite(val) and val >= 0 

650 except Exception: 

651 return False 

652 

653 def _safe_decrement_half_open_attempts(self) -> None: 

654 """Safely decrement half-open attempts.""" 

655 if self._state.state is not CircuitState.HALF_OPEN: 

656 return 

657 

658 if not self._is_valid_half_open_attempts(): 

659 # Reset if state is corrupted to prevent crash 

660 self._state.half_open_attempts = 0 

661 return 

662 

663 if self._state.half_open_attempts > 0: 

664 self._state.half_open_attempts -= 1 

665 

666 def _decrement_half_open(self, is_half_open: bool) -> None: 

667 """Decrement half-open attempt count if applicable. 

668 

669 Args: 

670 is_half_open: Whether the circuit was half-open before attempt. 

671 

672 """ 

673 if is_half_open: 

674 try: 

675 acquired = self._state.lock.acquire(timeout=0.1) 

676 except Exception: 

677 acquired = False 

678 

679 if not acquired: 

680 return 

681 try: 

682 self._safe_decrement_half_open_attempts() 

683 finally: 

684 self._safe_release() 

685 

686 def _handle_exception_in_call(self, e: Exception) -> None: 

687 try: 

688 is_failure = isinstance(e, self.config.failure_exceptions) 

689 except TypeError: 

690 is_failure = True 

691 if is_failure: 

692 self._record_failure(e) 

693 

694 def _execute_async_wrapper( 

695 self, 

696 func_coro: Callable[P, Awaitable[R]], 

697 ) -> Callable[P, Awaitable[R]]: 

698 @functools.wraps(func_coro) 

699 async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> R: 

700 if not self._should_attempt(): 

701 raise CircuitBreakerError( 

702 f"Circuit {self.name} is open", 

703 state=self._state.state, 

704 ) 

705 

706 is_half_open = self._state.state is CircuitState.HALF_OPEN 

707 

708 try: 

709 result = await func_coro(*args, **kwargs) 

710 return self._process_result(result) 

711 except Exception as e: 

712 self._handle_exception_in_call(e) 

713 raise 

714 finally: 

715 self._decrement_half_open(is_half_open) 

716 

717 return async_wrapper # type: ignore[misc] 

718 

719 def _execute_sync_wrapper( 

720 self, 

721 func_sync: Callable[P, R], 

722 ) -> Callable[P, R]: 

723 @functools.wraps(func_sync) 

724 def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: 

725 if not self._should_attempt(): 

726 raise CircuitBreakerError( 

727 f"Circuit {self.name} is open", 

728 state=self._state.state, 

729 ) 

730 

731 is_half_open = self._state.state is CircuitState.HALF_OPEN 

732 

733 try: 

734 result = func_sync(*args, **kwargs) 

735 return self._process_result(result) 

736 except Exception as e: 

737 self._handle_exception_in_call(e) 

738 raise 

739 finally: 

740 self._decrement_half_open(is_half_open) 

741 

742 return wrapper 

743 

744 def __call__( 

745 self, 

746 func: Callable[P, R] | Callable[P, Awaitable[R]], 

747 ) -> Callable[P, R] | Callable[P, Awaitable[R]]: 

748 """Decorate a sync or async function with circuit breaker protection.""" 

749 if inspect.iscoroutinefunction(func): 

750 func_coro = cast(Callable[P, Awaitable[R]], func) 

751 return self._execute_async_wrapper(func_coro) 

752 

753 func_sync = cast(Callable[P, R], func) 

754 return self._execute_sync_wrapper(func_sync) 

755 

756 

757def circuit_breaker( 

758 *, 

759 failure_threshold: int = 5, 

760 success_threshold: int = 2, 

761 timeout: float = 30.0, 

762 excluded_exceptions: tuple[type[Exception], ...] = (), 

763 failure_exceptions: tuple[type[Exception], ...] = (Exception,), 

764 name: str | None = None, 

765 on_state_change: Callable[[CircuitState, CircuitState], None] | None = None, 

766) -> CircuitBreakerDecorator: 

767 """Decorate a sync or async function with circuit breaker pattern. 

768 

769 Args: 

770 failure_threshold: Failures before opening circuit. 

771 success_threshold: Successes to close from half-open. 

772 timeout: Seconds before attempting half-open. 

773 excluded_exceptions: Exceptions that don't trip circuit. 

774 failure_exceptions: Exceptions that count as failures. 

775 name: Optional name for the circuit. 

776 on_state_change: Optional callback invoked on state transitions 

777 with (old_state, new_state). 

778 

779 Returns: 

780 Decorated function with circuit breaker protection. 

781 

782 Example: 

783 >>> @circuit_breaker(failure_threshold=3, timeout=60) 

784 ... def call_api(endpoint: str) -> dict: 

785 ... return requests.get(endpoint, timeout=10).json() 

786 

787 >>> @circuit_breaker( 

788 ... failure_threshold=3, 

789 ... on_state_change=lambda old, new: print(f"{old} -> {new}"), 

790 ... ) 

791 ... def monitored_call() -> str: 

792 ... return service.call() 

793 

794 """ 

795 

796 def decorator( 

797 func: Callable[P, R] | Callable[P, Awaitable[R]], 

798 ) -> Callable[P, R] | Callable[P, Awaitable[R]]: 

799 breaker = CircuitBreaker( 

800 failure_threshold=failure_threshold, 

801 success_threshold=success_threshold, 

802 timeout=timeout, 

803 excluded_exceptions=excluded_exceptions, 

804 failure_exceptions=failure_exceptions, 

805 name=name or cast(str, getattr(func, "__name__", "unknown")), 

806 on_state_change=on_state_change, 

807 ) 

808 return breaker(func) 

809 

810 return cast(CircuitBreakerDecorator, decorator)