tessl/pypi-pybreaker

Python implementation of the Circuit Breaker pattern for handling failing subsystems gracefully

—

Pending

Overview

Eval results

Files

Storage Backends

Name: tessl/pypi-pybreaker
Author: tessl

PyBreaker supports different storage backends for persisting circuit breaker state. This enables circuit breakers to maintain state across application restarts and coordinate state in distributed systems.

Capabilities

Memory Storage

In-memory storage for single-process applications where circuit state doesn't need to persist across restarts.

class CircuitMemoryStorage:
    def __init__(self, state: str):
        """
        Create a new in-memory storage instance.
        
        Args:
            state (str): Initial circuit state ('open', 'closed', 'half-open')
        """

    @property
    def state(self) -> str:
        """Current circuit breaker state."""

    @state.setter
    def state(self, state: str) -> None:
        """Set the current circuit breaker state."""

    @property
    def counter(self) -> int:
        """Current failure counter value."""

    @property
    def success_counter(self) -> int:
        """Current success counter value."""

    @property
    def opened_at(self) -> datetime | None:
        """Datetime when circuit was last opened."""

    @opened_at.setter
    def opened_at(self, datetime: datetime) -> None:
        """Set when circuit was opened."""

    def increment_counter(self) -> None:
        """Increment the failure counter by one."""

    def reset_counter(self) -> None:
        """Reset the failure counter to zero."""

    def increment_success_counter(self) -> None:
        """Increment the success counter by one."""

    def reset_success_counter(self) -> None:
        """Reset the success counter to zero."""

Redis Storage

Redis-backed storage for distributed applications where circuit state needs to be shared across multiple processes or servers.

class CircuitRedisStorage:
    def __init__(self, 
                 state: str,
                 redis_object: Redis,
                 namespace: str | None = None,
                 fallback_circuit_state: str = "closed",
                 cluster_mode: bool = False):
        """
        Create a new Redis storage instance.
        
        Args:
            state (str): Initial circuit state
            redis_object (Redis): Redis client instance (e.g., redis.Redis())
            namespace (str | None): Optional namespace for Redis keys
            fallback_circuit_state (str): State to use when Redis is unavailable
            cluster_mode (bool): Enable Redis cluster mode support
            
        Note:
            Requires 'redis' package to be installed.
            Do not use decode_responses=True in Redis client.
            Check pybreaker.HAS_REDIS_SUPPORT to verify availability.
        """

    @property
    def state(self) -> str:
        """
        Current circuit breaker state from Redis.
        Falls back to fallback_circuit_state on Redis errors.
        """

    @state.setter
    def state(self, state: str) -> None:
        """Set the current circuit breaker state in Redis."""

    @property
    def counter(self) -> int:
        """Current failure counter value from Redis."""

    @property
    def success_counter(self) -> int:
        """Current success counter value from Redis."""

    @property
    def opened_at(self) -> datetime | None:
        """Datetime when circuit was last opened, from Redis."""

    @opened_at.setter
    def opened_at(self, datetime: datetime) -> None:
        """Atomically set when circuit was opened in Redis."""

    def increment_counter(self) -> None:
        """Increment the failure counter in Redis."""

    def reset_counter(self) -> None:
        """Reset the failure counter to zero in Redis."""

    def increment_success_counter(self) -> None:
        """Increment the success counter in Redis."""

    def reset_success_counter(self) -> None:
        """Reset the success counter to zero in Redis."""

Storage Base Class

Abstract base class defining the storage interface for custom storage implementations. All concrete storage classes must inherit from this class and implement the abstract methods.

class CircuitBreakerStorage:
    def __init__(self, name: str) -> None:
        """
        Create a new storage instance.
        
        Args:
            name (str): Human-friendly name for this storage type
        """

    @property
    def name(self) -> str:
        """Human friendly name that identifies this storage type."""

    @property
    def state(self) -> str:
        """Override this method to retrieve the current circuit breaker state."""

    @state.setter  
    def state(self, state: str) -> None:
        """Override this method to set the current circuit breaker state."""

    def increment_counter(self) -> None:
        """Override this method to increase the failure counter by one."""

    def reset_counter(self) -> None:
        """Override this method to set the failure counter to zero."""

    def increment_success_counter(self) -> None:
        """Override this method to increase the success counter by one."""

    def reset_success_counter(self) -> None:
        """Override this method to set the success counter to zero."""

    @property
    def counter(self) -> int:
        """Override this method to retrieve the current failure counter value."""

    @property
    def success_counter(self) -> int:
        """Override this method to retrieve the current success counter value."""

    @property
    def opened_at(self) -> datetime | None:
        """Override this method to retrieve when the circuit was opened."""

    @opened_at.setter
    def opened_at(self, datetime: datetime) -> None:
        """Override this method to set when the circuit was opened."""

Usage Examples

Memory Storage (Default)

import pybreaker

# Memory storage is used by default
breaker = pybreaker.CircuitBreaker(fail_max=5, reset_timeout=60)

# Explicitly specify memory storage  
storage = pybreaker.CircuitMemoryStorage(state=pybreaker.STATE_CLOSED)
breaker = pybreaker.CircuitBreaker(state_storage=storage)

Redis Storage for Distributed Systems

import pybreaker
import redis

# Create Redis client
redis_client = redis.Redis(host='localhost', port=6379, db=0)

# Create Redis storage
storage = pybreaker.CircuitRedisStorage(
    state=pybreaker.STATE_CLOSED,
    redis_object=redis_client,
    namespace="myapp",  # Optional namespace for keys
    fallback_circuit_state=pybreaker.STATE_CLOSED
)

# Create circuit breaker with Redis storage
breaker = pybreaker.CircuitBreaker(
    fail_max=5,
    reset_timeout=60,
    state_storage=storage,
    name="user_service"
)

Redis Storage with Django

import pybreaker
from django_redis import get_redis_connection

# Use existing Django Redis connection
redis_client = get_redis_connection('default')

storage = pybreaker.CircuitRedisStorage(
    state=pybreaker.STATE_CLOSED,
    redis_object=redis_client,
    namespace="circuit_breakers"
)

breaker = pybreaker.CircuitBreaker(state_storage=storage)

Multiple Circuit Breakers with Redis

import pybreaker
import redis

redis_client = redis.Redis(host='localhost', port=6379, db=0)

# Each service gets its own namespace
user_storage = pybreaker.CircuitRedisStorage(
    state=pybreaker.STATE_CLOSED,
    redis_object=redis_client,
    namespace="user_service"
)

payment_storage = pybreaker.CircuitRedisStorage(
    state=pybreaker.STATE_CLOSED,
    redis_object=redis_client,
    namespace="payment_service"
)

user_breaker = pybreaker.CircuitBreaker(state_storage=user_storage)
payment_breaker = pybreaker.CircuitBreaker(state_storage=payment_storage)

Redis Cluster Mode

import pybreaker
import redis

# For Redis cluster deployments
redis_client = redis.Redis(host='cluster-endpoint', port=6379)

storage = pybreaker.CircuitRedisStorage(
    state=pybreaker.STATE_CLOSED,
    redis_object=redis_client,
    cluster_mode=True  # Enable cluster mode
)

breaker = pybreaker.CircuitBreaker(state_storage=storage)

Custom Storage Implementation

import pybreaker
import sqlite3
from datetime import datetime

class SQLiteStorage(pybreaker.CircuitBreakerStorage):
    def __init__(self, db_path, circuit_name):
        super().__init__("sqlite")
        self.db_path = db_path
        self.circuit_name = circuit_name
        self._init_db()
    
    def _init_db(self):
        # Initialize SQLite database and tables
        pass
    
    @property
    def state(self):
        # Implement state retrieval from SQLite
        pass
    
    @state.setter
    def state(self, state):
        # Implement state storage to SQLite
        pass
    
    # Implement other required methods...

# Use custom storage
storage = SQLiteStorage("/path/to/circuit.db", "user_service")
breaker = pybreaker.CircuitBreaker(state_storage=storage)

Install with Tessl CLI