Circuit Breaker

Circuit breaker plugin for resilient external service calls.

Classes

class arshai.plugins.resilience.circuit_breaker.CircuitBreakerConfig(failure_threshold=5, recovery_timeout=60, expected_exception=<class 'Exception'>, name=None)

Bases: object

Configuration for circuit breaker.

failure_threshold: int = 5

recovery_timeout: int = 60

expected_exception

alias of Exception

name: Optional[str] = None

__init__(failure_threshold=5, recovery_timeout=60, expected_exception=<class 'Exception'>, name=None)

class arshai.plugins.resilience.circuit_breaker.CircuitBreakerError

Bases: Exception

Raised when circuit is open.

class arshai.plugins.resilience.circuit_breaker.CircuitBreakerPlugin(config=None, enable_metrics=True)

Bases: Plugin

Circuit breaker plugin for resilient external service calls.

This is an OPTIONAL plugin that adds resilience patterns without affecting existing Arshai functionality.

Example

# Optional usage - existing code continues to work cb = CircuitBreakerPlugin(config)

# Wrap async calls result = await cb.call_async(external_service.fetch_data, param1, param2)

# Wrap sync calls result = cb.call(external_service.fetch_data_sync, param1, param2)

# Decorator pattern (opt-in) @cb.protected async def my_external_call():

return await external_api.request()

name = 'circuit_breaker'

version = '1.0.0'

__init__(config=None, enable_metrics=True)

Initialize circuit breaker with optional config.

Parameters:

• config (Optional[CircuitBreakerConfig]) – Optional configuration. Defaults to sensible values.

• enable_metrics (bool) – Whether to enable performance metrics collection.

property state: CircuitState

Get current circuit state.

async call_async(func, *args, **kwargs)

Execute async function with circuit breaker protection.

Parameters:

• func (Callable[..., TypeVar(T)]) – Async function to call

• *args – Positional arguments for func

• **kwargs – Keyword arguments for func

Return type:

TypeVar(T)

Returns:

Function result

Raises:

• CircuitBreakerError – If circuit is open

• Exception – If function fails

call(func, *args, **kwargs)

Execute sync function with circuit breaker protection.

Parameters:

• func (Callable[..., TypeVar(T)]) – Sync function to call

• *args – Positional arguments for func

• **kwargs – Keyword arguments for func

Return type:

TypeVar(T)

Returns:

Function result

Raises:

• CircuitBreakerError – If circuit is open

• Exception – If function fails

protected(func)

Decorator to protect functions with circuit breaker.

Works with both sync and async functions.

Return type:

Callable

Example

@circuit_breaker.protected async def external_call():

return await api.request()

reset()

Manually reset the circuit breaker.

Return type:

None

get_stats()

Get circuit breaker statistics including performance metrics.

Return type:

dict

get_metadata()

Get plugin metadata (required by Plugin base class).

Return type:

dict

initialize()

Initialize the plugin (required by Plugin base class).

Return type:

None

shutdown()

Shutdown the plugin (required by Plugin base class).

Return type:

None

class arshai.plugins.resilience.circuit_breaker.CircuitState(value)

Bases: Enum

Circuit breaker states.

CLOSED = 'closed'

OPEN = 'open'

HALF_OPEN = 'half_open'