proxywhirl.circuit_breaker.sync

Circuit breaker pattern implementation for proxy failure management.

IMPORTANT: This module provides SYNCHRONOUS circuit breaker implementation.

Usage Guidelines:

• CircuitBreaker: Use in synchronous contexts only (threading.Lock)

• AsyncCircuitBreaker: Use in async contexts (from circuit_breaker_async module)

Example

Synchronous usage:

>>> cb = CircuitBreaker(proxy_id="proxy-1")
>>> cb.record_failure()  # Thread-safe
>>> if cb.should_attempt_request():
...     # make request
...     cb.record_success()

Async usage (use AsyncCircuitBreaker instead):

>>> from proxywhirl.circuit_breaker_async import AsyncCircuitBreaker
>>> cb = AsyncCircuitBreaker(proxy_id="proxy-1")
>>> await cb.record_failure()  # Event loop safe
>>> if await cb.should_attempt_request():
...     # make async request
...     await cb.record_success()

Classes

CircuitBreaker

Circuit breaker for a single proxy (SYNCHRONOUS implementation).

Module Contents

class proxywhirl.circuit_breaker.sync.CircuitBreaker(/, **data)

Bases: proxywhirl.circuit_breaker.base.CircuitBreakerBase

Circuit breaker for a single proxy (SYNCHRONOUS implementation).

WARNING: This class uses threading.Lock and is designed for SYNCHRONOUS contexts only. For async applications, use AsyncCircuitBreaker from circuit_breaker_async module.

Parameters:

data (Any)

proxy_id

Unique identifier for the proxy

state

Current circuit breaker state (CLOSED, OPEN, HALF_OPEN)

failure_count

Number of failures in current window

failure_threshold

Number of failures before opening circuit

window_duration

Rolling window duration in seconds

timeout_duration

How long circuit stays open before testing recovery

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

classmethod create(proxy_id, config=None, **kwargs)

Factory method to create a circuit breaker with optional config.

Parameters:

• proxy_id (str) – Unique identifier for the proxy

• config (proxywhirl.models.CircuitBreakerConfig | None) – CircuitBreakerConfig with settings

• **kwargs (Any) – Additional CircuitBreaker field overrides

Returns:

CircuitBreaker instance

Return type:

CircuitBreaker

classmethod from_config(proxy_id, config=None, **kwargs)

Backward-compatible alias for create().

Parameters:

• proxy_id (str)

• config (proxywhirl.models.CircuitBreakerConfig | None)

• kwargs (Any)

Return type:

CircuitBreaker

record_failure()

Record a failure and update state if threshold reached.

Thread-safe - acquires threading.Lock for the entire operation.

Return type:

None

record_success()

Record a success and potentially close circuit.

Thread-safe - acquires threading.Lock for the entire operation.

Return type:

None

reset()

Manually reset circuit breaker to CLOSED state.

Thread-safe - acquires threading.Lock before resetting state.

Return type:

None

should_attempt_request()

Check if proxy is available for requests.

Thread-safe - acquires threading.Lock for the entire operation.

Returns:

True if proxy should be attempted, False if circuit is open.

Return type:

bool