(self)
| 543 | |
| 544 | @property |
| 545 | def state(self) -> CircuitState: |
| 546 | with self._lock: |
| 547 | if self._state == CircuitState.OPEN: |
| 548 | elapsed = time.monotonic() - (self._last_failure_time or 0) |
| 549 | if elapsed >= self.recovery_seconds: |
| 550 | self._state = CircuitState.HALF_OPEN |
| 551 | log.info("circuit_breaker.half_open", recovery_seconds=self.recovery_seconds) |
| 552 | return self._state |
| 553 | |
| 554 | def is_open(self) -> bool: |
| 555 | return self.state == CircuitState.OPEN |
nothing calls this directly
no outgoing calls
no test coverage detected