Core¶

이 페이지는 사용자에게 노출되는 주요 클래스와 예외를 문서화합니다. 이는 Fluxgate 사용을 위한 주요 진입점입니다.

Circuit Breaker 클래스¶

이것은 Circuit Breaker를 생성하기 위해 인스턴스화할 주요 클래스입니다.

fluxgate.circuitbreaker.CircuitBreaker ¶

CircuitBreaker(
    name: str,
    window: IWindow | None = None,
    tracker: ITracker | None = None,
    tripper: ITripper | None = None,
    retry: IRetry | None = None,
    permit: IPermit | None = None,
    slow_threshold: float = 60.0,
    listeners: Iterable[IListener] = (),
)

Synchronous circuit breaker implementation.

Protects your service from cascading failures by monitoring call failures and temporarily blocking calls when a failure threshold is reached.

The circuit breaker operates in three main states:

CLOSED: Normal operation, calls pass through
OPEN: Failure threshold exceeded, calls are blocked
HALF_OPEN: Testing if the service recovered, limited calls allowed

Parameters:

Name	Type	Description	Default
`name`	`str`	Circuit breaker identifier	required
`window`	`IWindow \| None`	Sliding window for metrics collection (default: CountWindow(100))	`None`
`tracker`	`ITracker \| None`	Determines which exceptions to track as failures (default: All())	`None`
`tripper`	`ITripper \| None`	Condition to open/close the circuit based on metrics (default: MinRequests(100) & (FailureRate(0.5) \| SlowRate(1.0)))	`None`
`retry`	`IRetry \| None`	Strategy for transitioning from OPEN to HALF_OPEN (default: Cooldown(60.0))	`None`
`permit`	`IPermit \| None`	Strategy for allowing calls in HALF_OPEN state (default: RampUp(0.0, 1.0, 60.0))	`None`
`slow_threshold`	`float`	Duration threshold in seconds to mark calls as slow (default: 60.0)	`60.0`
`listeners`	`Iterable[IListener]`	Event listeners for state transitions (default: empty)	`()`

Examples:

Basic usage with defaults:

>>> cb = CircuitBreaker("my-service")
>>> @cb
... def call_api():
...     return requests.get("https://api.example.com")

Custom configuration:

>>> cb = CircuitBreaker(
...     name="payment_api",
...     tracker=TypeOf(ConnectionError),
...     tripper=MinRequests(10) & FailureRate(0.5),
...     slow_threshold=1.0,
... )

Note

This implementation is NOT thread-safe. Each process maintains its own independent circuit breaker state. For asyncio applications, use AsyncCircuitBreaker instead.

Source code in fluxgate/circuitbreaker.py

def __init__(
    self,
    name: str,
    window: IWindow | None = None,
    tracker: ITracker | None = None,
    tripper: ITripper | None = None,
    retry: IRetry | None = None,
    permit: IPermit | None = None,
    slow_threshold: float = 60.0,
    listeners: Iterable[IListener] = (),
) -> None:
    self._name = name
    self._window = window or CountWindow(100)
    self._tracker = tracker or All()
    self._tripper = tripper or MinRequests(100) & (FailureRate(0.5) | SlowRate(1.0))
    self._retry = retry or Cooldown(60.0)
    self._permit = permit or RampUp(0.0, 1.0, 60.0)
    self._listeners = tuple(listeners)
    self._slow_threshold = slow_threshold
    self._changed_at = time.time()
    self._reopens = 0
    self._consecutive_failures = 0
    self._state: CircuitBreaker._State = self._Closed(self)

call ¶

__call__(func: Callable[P, R]) -> Callable[P, R]

__call__(
    func: None = None,
    *,
    fallback: Callable[[Exception], R] | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]

__call__(
    func: Callable[P, R] | None = None,
    *,
    fallback: Callable[[Exception], R] | None = None,
) -> (
    Callable[P, R]
    | Callable[[Callable[P, R]], Callable[P, R]]
)

Decorate a function with circuit breaker protection.

Examples:

>>> @cb
... def api_call():
...     return requests.get("https://api.example.com")

>>> @cb(fallback=lambda e: cached_value)
... def api_call():
...     return requests.get("https://api.example.com")

Parameters:

Name	Type	Description	Default
`func`	`Callable[P, R] \| None`	Function to protect	`None`
`fallback`	`Callable[[Exception], R] \| None`	Optional function to call on exception. Receives the exception as argument and should return a fallback value or re-raise.	`None`

Returns:

Type	Description
`Callable[P, R] \| Callable[[Callable[P, R]], Callable[P, R]]`	Wrapped function with circuit breaker behavior

Raises:

Type	Description
`CallNotPermittedError`	When circuit is OPEN or FORCED_OPEN (if no fallback)

Source code in fluxgate/circuitbreaker.py

def __call__(
    self,
    func: Callable[P, R] | None = None,
    *,
    fallback: Callable[[Exception], R] | None = None,
) -> Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]:
    """Decorate a function with circuit breaker protection.

    Examples:
        >>> @cb
        ... def api_call():
        ...     return requests.get("https://api.example.com")

        >>> @cb(fallback=lambda e: cached_value)
        ... def api_call():
        ...     return requests.get("https://api.example.com")

    Args:
        func: Function to protect
        fallback: Optional function to call on exception. Receives the exception
            as argument and should return a fallback value or re-raise.

    Returns:
        Wrapped function with circuit breaker behavior

    Raises:
        CallNotPermittedError: When circuit is OPEN or FORCED_OPEN (if no fallback)
    """

    def decorator(f: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(f)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            try:
                return self._state.execute(f, *args, **kwargs)
            except Exception as e:
                if fallback is not None:
                    return fallback(e)
                raise

        return wrapper

    if func is not None:
        return decorator(func)
    return decorator

call ¶

call(
    func: Callable[P, R], /, *args: args, **kwargs: kwargs
) -> R

Execute a function with circuit breaker protection.

Examples:

>>> cb.call(requests.get, "https://api.example.com")

Parameters:

Name	Type	Description	Default
`func`	`Callable[P, R]`	Function to execute	required
`*args`	`args`	Positional arguments for the function	`()`
`**kwargs`	`kwargs`	Keyword arguments for the function	`{}`

Returns:

Type	Description
`R`	Function result

Raises:

Type	Description
`CallNotPermittedError`	When circuit is OPEN or FORCED_OPEN

Source code in fluxgate/circuitbreaker.py

def call(
    self,
    func: Callable[P, R],
    /,
    *args: P.args,
    **kwargs: P.kwargs,
) -> R:
    """Execute a function with circuit breaker protection.

    Examples:
        >>> cb.call(requests.get, "https://api.example.com")

    Args:
        func: Function to execute
        *args: Positional arguments for the function
        **kwargs: Keyword arguments for the function

    Returns:
        Function result

    Raises:
        CallNotPermittedError: When circuit is OPEN or FORCED_OPEN
    """
    return self._state.execute(func, *args, **kwargs)

call_with_fallback ¶

call_with_fallback(
    func: Callable[P, R],
    fallback: Callable[[Exception], R],
    /,
    *args: args,
    **kwargs: kwargs,
) -> R

Execute a function with circuit breaker protection and fallback.

Examples:

>>> cb.call_with_fallback(fetch_data, lambda e: cached_data)

Parameters:

Name	Type	Description	Default
`func`	`Callable[P, R]`	Function to execute	required
`fallback`	`Callable[[Exception], R]`	Function to call on exception. Receives the exception as argument and should return a fallback value or re-raise.	required
`*args`	`args`	Positional arguments for the function	`()`
`**kwargs`	`kwargs`	Keyword arguments for the function	`{}`

Returns:

Type	Description
`R`	Function result or fallback result

Source code in fluxgate/circuitbreaker.py

def call_with_fallback(
    self,
    func: Callable[P, R],
    fallback: Callable[[Exception], R],
    /,
    *args: P.args,
    **kwargs: P.kwargs,
) -> R:
    """Execute a function with circuit breaker protection and fallback.

    Examples:
        >>> cb.call_with_fallback(fetch_data, lambda e: cached_data)

    Args:
        func: Function to execute
        fallback: Function to call on exception. Receives the exception
            as argument and should return a fallback value or re-raise.
        *args: Positional arguments for the function
        **kwargs: Keyword arguments for the function

    Returns:
        Function result or fallback result
    """
    try:
        return self._state.execute(func, *args, **kwargs)
    except Exception as e:
        return fallback(e)

info ¶

info() -> CircuitBreakerInfo

Get current circuit breaker state and metrics.

Returns:

Type	Description
`CircuitBreakerInfo`	Dictionary with circuit breaker state information

Source code in fluxgate/circuitbreaker.py

def info(self) -> CircuitBreakerInfo:
    """Get current circuit breaker state and metrics.

    Returns:
        Dictionary with circuit breaker state information
    """
    return CircuitBreakerInfo(
        name=self._name,
        state=self._state.get_state_enum().value,
        changed_at=self._changed_at,
        reopens=self._reopens,
        metrics=self._window.get_metric(),
    )

reset ¶

reset() -> None

Reset circuit breaker to CLOSED state and clear metrics.

Source code in fluxgate/circuitbreaker.py

def reset(self) -> None:
    """Reset circuit breaker to CLOSED state and clear metrics."""
    current_state = self._state.get_state_enum()
    self._state = self._Closed(self)
    self._changed_at = time.time()
    self._reopens = 0
    self._consecutive_failures = 0
    self._window.reset()

    signal = Signal(
        circuit_name=self._name,
        old_state=current_state,
        new_state=StateEnum.CLOSED,
        timestamp=self._changed_at,
    )
    self._notify(signal)

disable ¶

disable() -> None

Disable circuit breaker (all calls pass through without tracking).

Source code in fluxgate/circuitbreaker.py

def disable(self) -> None:
    """Disable circuit breaker (all calls pass through without tracking)."""
    self._transition_to(StateEnum.DISABLED)

metrics_only ¶

metrics_only() -> None

Enable metrics-only mode (track metrics but never open circuit).

Source code in fluxgate/circuitbreaker.py

def metrics_only(self) -> None:
    """Enable metrics-only mode (track metrics but never open circuit)."""
    self._transition_to(StateEnum.METRICS_ONLY)

force_open ¶

force_open() -> None

Force circuit breaker to OPEN state (all calls blocked).

Source code in fluxgate/circuitbreaker.py

def force_open(self) -> None:
    """Force circuit breaker to OPEN state (all calls blocked)."""
    self._transition_to(StateEnum.FORCED_OPEN)

fluxgate.circuitbreaker.AsyncCircuitBreaker ¶

AsyncCircuitBreaker(
    name: str,
    window: IWindow | None = None,
    tracker: ITracker | None = None,
    tripper: ITripper | None = None,
    retry: IRetry | None = None,
    permit: IPermit | None = None,
    slow_threshold: float = 60.0,
    max_half_open_calls: int = 10,
    listeners: Iterable[IListener | IAsyncListener] = (),
)

Asynchronous circuit breaker implementation for asyncio applications.

Thread-safe circuit breaker with async/await support. Uses asyncio locks to coordinate state transitions and metric updates across concurrent tasks.

The circuit breaker operates in three main states:

CLOSED: Normal operation, calls pass through
OPEN: Failure threshold exceeded, calls are blocked
HALF_OPEN: Testing if the service recovered, limited calls allowed

Parameters:

Name	Type	Description	Default
`name`	`str`	Circuit breaker identifier	required
`window`	`IWindow \| None`	Sliding window for metrics collection (default: CountWindow(100))	`None`
`tracker`	`ITracker \| None`	Determines which exceptions to track as failures (default: All())	`None`
`tripper`	`ITripper \| None`	Condition to open/close the circuit based on metrics (default: MinRequests(100) & (FailureRate(0.5) \| SlowRate(1.0)))	`None`
`retry`	`IRetry \| None`	Strategy for transitioning from OPEN to HALF_OPEN (default: Cooldown(60.0))	`None`
`permit`	`IPermit \| None`	Strategy for allowing calls in HALF_OPEN state (default: RampUp(0.0, 1.0, 60.0))	`None`
`slow_threshold`	`float`	Duration threshold in seconds to mark calls as slow (default: 60.0)	`60.0`
`max_half_open_calls`	`int`	Maximum concurrent calls allowed in HALF_OPEN state (default: 10)	`10`
`listeners`	`Iterable[IListener \| IAsyncListener]`	Event listeners for state transitions (default: empty)	`()`

Examples:

Basic usage with defaults:

>>> cb = AsyncCircuitBreaker("my-service")
>>> @cb
... async def call_api():
...     async with httpx.AsyncClient() as client:
...         return await client.get("https://api.example.com")

Custom configuration:

>>> cb = AsyncCircuitBreaker(
...     name="payment_api",
...     tracker=TypeOf(httpx.ConnectError),
...     tripper=MinRequests(10) & FailureRate(0.5),
...     slow_threshold=1.0,
... )

Note

Uses asyncio locks for thread safety within a single event loop. Each process maintains its own independent circuit breaker state.

Source code in fluxgate/circuitbreaker.py

def __init__(
    self,
    name: str,
    window: IWindow | None = None,
    tracker: ITracker | None = None,
    tripper: ITripper | None = None,
    retry: IRetry | None = None,
    permit: IPermit | None = None,
    slow_threshold: float = 60.0,
    max_half_open_calls: int = 10,
    listeners: Iterable[IListener | IAsyncListener] = (),
) -> None:
    self._name = name
    self._window = window or CountWindow(100)
    self._tracker = tracker or All()
    self._tripper = tripper or MinRequests(100) & (FailureRate(0.5) | SlowRate(1.0))
    self._retry = retry or Cooldown(60.0)
    self._permit = permit or RampUp(0.0, 1.0, 60.0)
    self._listeners = tuple(listeners)
    self._slow_threshold = slow_threshold
    self._changed_at = time.time()
    self._reopens = 0
    self._consecutive_failures = 0
    self._state: AsyncCircuitBreaker._State = self._Closed(self)
    self._state_lock = asyncio.Lock()
    self._half_open_semaphore = asyncio.Semaphore(max_half_open_calls)
    self._window_lock = asyncio.Lock()

call ¶

__call__(
    func: Callable[P, Awaitable[R]],
) -> Callable[P, Awaitable[R]]

__call__(
    func: None = None,
    *,
    fallback: Callable[[Exception], R] | None = None,
) -> Callable[
    [Callable[P, Awaitable[R]]],
    Callable[P, Awaitable[R]],
]

__call__(
    func: Callable[P, Awaitable[R]] | None = None,
    *,
    fallback: Callable[[Exception], R] | None = None,
) -> (
    Callable[P, Awaitable[R]]
    | Callable[
        [Callable[P, Awaitable[R]]],
        Callable[P, Awaitable[R]],
    ]
)

Decorate an awaitable function with circuit breaker protection.

Examples:

>>> @cb
... async def api_call():
...     async with httpx.AsyncClient() as client:
...         return await client.get("https://api.example.com")

>>> @cb(fallback=lambda e: cached_value)
... async def api_call():
...     return await fetch_data()

Parameters:

Name	Type	Description	Default
`func`	`Callable[P, Awaitable[R]] \| None`	Awaitable function to protect	`None`
`fallback`	`Callable[[Exception], R] \| None`	Optional function to call on exception. Receives the exception as argument and should return a fallback value or re-raise.	`None`

Returns:

Type	Description
`Callable[P, Awaitable[R]] \| Callable[[Callable[P, Awaitable[R]]], Callable[P, Awaitable[R]]]`	Wrapped awaitable function with circuit breaker behavior

Raises:

Type	Description
`CallNotPermittedError`	When circuit is OPEN or FORCED_OPEN (if no fallback)

Source code in fluxgate/circuitbreaker.py

def __call__(
    self,
    func: Callable[P, Awaitable[R]] | None = None,
    *,
    fallback: Callable[[Exception], R] | None = None,
) -> (
    Callable[P, Awaitable[R]]
    | Callable[[Callable[P, Awaitable[R]]], Callable[P, Awaitable[R]]]
):
    """Decorate an awaitable function with circuit breaker protection.

    Examples:
        >>> @cb
        ... async def api_call():
        ...     async with httpx.AsyncClient() as client:
        ...         return await client.get("https://api.example.com")

        >>> @cb(fallback=lambda e: cached_value)
        ... async def api_call():
        ...     return await fetch_data()

    Args:
        func: Awaitable function to protect
        fallback: Optional function to call on exception. Receives the exception
            as argument and should return a fallback value or re-raise.

    Returns:
        Wrapped awaitable function with circuit breaker behavior

    Raises:
        CallNotPermittedError: When circuit is OPEN or FORCED_OPEN (if no fallback)
    """

    def decorator(
        f: Callable[P, Awaitable[R]],
    ) -> Callable[P, Awaitable[R]]:
        @functools.wraps(f)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            try:
                return await self._state.execute(f, *args, **kwargs)
            except Exception as e:
                if fallback is not None:
                    return fallback(e)
                raise

        return wrapper

    if func is not None:
        return decorator(func)
    return decorator

call `async` ¶

call(
    func: Callable[P, Awaitable[R]],
    /,
    *args: args,
    **kwargs: kwargs,
) -> R

Execute an awaitable function with circuit breaker protection.

Examples:

>>> await cb.call(client.get, "https://api.example.com")

Parameters:

Name	Type	Description	Default
`func`	`Callable[P, Awaitable[R]]`	Awaitable function to execute	required
`*args`	`args`	Positional arguments for the function	`()`
`**kwargs`	`kwargs`	Keyword arguments for the function	`{}`

Returns:

Type	Description
`R`	Function result

Raises:

Type	Description
`CallNotPermittedError`	When circuit is OPEN or FORCED_OPEN

Source code in fluxgate/circuitbreaker.py

async def call(
    self,
    func: Callable[P, Awaitable[R]],
    /,
    *args: P.args,
    **kwargs: P.kwargs,
) -> R:
    """Execute an awaitable function with circuit breaker protection.

    Examples:
        >>> await cb.call(client.get, "https://api.example.com")

    Args:
        func: Awaitable function to execute
        *args: Positional arguments for the function
        **kwargs: Keyword arguments for the function

    Returns:
        Function result

    Raises:
        CallNotPermittedError: When circuit is OPEN or FORCED_OPEN
    """
    return await self._state.execute(func, *args, **kwargs)

call_with_fallback `async` ¶

call_with_fallback(
    func: Callable[P, Awaitable[R]],
    fallback: Callable[[Exception], R],
    /,
    *args: args,
    **kwargs: kwargs,
) -> R

Execute an awaitable function with circuit breaker protection and fallback.

Examples:

>>> await cb.call_with_fallback(fetch_data, lambda e: cached_data)

Parameters:

Name	Type	Description	Default
`func`	`Callable[P, Awaitable[R]]`	Awaitable function to execute	required
`fallback`	`Callable[[Exception], R]`	Function to call on exception. Receives the exception as argument and should return a fallback value or re-raise.	required
`*args`	`args`	Positional arguments for the function	`()`
`**kwargs`	`kwargs`	Keyword arguments for the function	`{}`

Returns:

Type	Description
`R`	Function result or fallback result

Source code in fluxgate/circuitbreaker.py

async def call_with_fallback(
    self,
    func: Callable[P, Awaitable[R]],
    fallback: Callable[[Exception], R],
    /,
    *args: P.args,
    **kwargs: P.kwargs,
) -> R:
    """Execute an awaitable function with circuit breaker protection and fallback.

    Examples:
        >>> await cb.call_with_fallback(fetch_data, lambda e: cached_data)

    Args:
        func: Awaitable function to execute
        fallback: Function to call on exception. Receives the exception
            as argument and should return a fallback value or re-raise.
        *args: Positional arguments for the function
        **kwargs: Keyword arguments for the function

    Returns:
        Function result or fallback result
    """
    try:
        return await self._state.execute(func, *args, **kwargs)
    except Exception as e:
        return fallback(e)

info ¶

info() -> CircuitBreakerInfo

Get current circuit breaker state and metrics.

Returns:

Type	Description
`CircuitBreakerInfo`	Dictionary with circuit breaker state information

Source code in fluxgate/circuitbreaker.py

def info(self) -> CircuitBreakerInfo:
    """Get current circuit breaker state and metrics.

    Returns:
        Dictionary with circuit breaker state information
    """
    return CircuitBreakerInfo(
        name=self._name,
        state=self._state.get_state_enum().value,
        changed_at=self._changed_at,
        reopens=self._reopens,
        metrics=self._window.get_metric(),
    )

reset `async` ¶

reset() -> None

Reset circuit breaker to CLOSED state and clear metrics.

Source code in fluxgate/circuitbreaker.py

async def reset(self) -> None:
    """Reset circuit breaker to CLOSED state and clear metrics."""
    async with self._state_lock:
        await self._transition_to(StateEnum.CLOSED)

disable `async` ¶

disable() -> None

Disable circuit breaker (all calls pass through without tracking).

Source code in fluxgate/circuitbreaker.py

async def disable(self) -> None:
    """Disable circuit breaker (all calls pass through without tracking)."""
    async with self._state_lock:
        await self._transition_to(StateEnum.DISABLED)

metrics_only `async` ¶

metrics_only() -> None

Enable metrics-only mode (track metrics but never open circuit).

Source code in fluxgate/circuitbreaker.py

async def metrics_only(self) -> None:
    """Enable metrics-only mode (track metrics but never open circuit)."""
    async with self._state_lock:
        await self._transition_to(StateEnum.METRICS_ONLY)

force_open `async` ¶

force_open() -> None

Force circuit breaker to OPEN state (all calls blocked).

Source code in fluxgate/circuitbreaker.py

async def force_open(self) -> None:
    """Force circuit breaker to OPEN state (all calls blocked)."""
    async with self._state_lock:
        await self._transition_to(StateEnum.FORCED_OPEN)

상태 및 예외¶

이것은 가장 일반적으로 상호 작용할 열거형(enum) 및 예외입니다.

fluxgate.state.StateEnum ¶

Bases: Enum

CLOSED `class-attribute` `instance-attribute` ¶

CLOSED = 'closed'

OPEN `class-attribute` `instance-attribute` ¶

OPEN = 'open'

HALF_OPEN `class-attribute` `instance-attribute` ¶

HALF_OPEN = 'half_open'

METRICS_ONLY `class-attribute` `instance-attribute` ¶

METRICS_ONLY = 'metrics_only'

DISABLED `class-attribute` `instance-attribute` ¶

DISABLED = 'disabled'

FORCED_OPEN `class-attribute` `instance-attribute` ¶

FORCED_OPEN = 'forced_open'

fluxgate.errors.CallNotPermittedError ¶

CallNotPermittedError(message: str)

Bases: Exception

Raised when a call is not permitted due to circuit breaker state.

Source code in fluxgate/errors.py

def __init__(self, message: str) -> None:
    super().__init__(message)
    self.message = message

message `instance-attribute` ¶

message = message

Core¶

Circuit Breaker 클래스¶

fluxgate.circuitbreaker.CircuitBreaker ¶

__call__ ¶

call ¶

call_with_fallback ¶

info ¶

reset ¶

disable ¶

metrics_only ¶

force_open ¶

fluxgate.circuitbreaker.AsyncCircuitBreaker ¶

__call__ ¶

call async ¶

call_with_fallback async ¶

info ¶

reset async ¶

disable async ¶

metrics_only async ¶

force_open async ¶

상태 및 예외¶

fluxgate.state.StateEnum ¶

CLOSED class-attribute instance-attribute ¶

OPEN class-attribute instance-attribute ¶

HALF_OPEN class-attribute instance-attribute ¶

METRICS_ONLY class-attribute instance-attribute ¶

DISABLED class-attribute instance-attribute ¶

FORCED_OPEN class-attribute instance-attribute ¶

fluxgate.errors.CallNotPermittedError ¶

message instance-attribute ¶

call ¶

call ¶

call `async` ¶

call_with_fallback `async` ¶

reset `async` ¶

disable `async` ¶

metrics_only `async` ¶

force_open `async` ¶

CLOSED `class-attribute` `instance-attribute` ¶

OPEN `class-attribute` `instance-attribute` ¶

HALF_OPEN `class-attribute` `instance-attribute` ¶

METRICS_ONLY `class-attribute` `instance-attribute` ¶

DISABLED `class-attribute` `instance-attribute` ¶

FORCED_OPEN `class-attribute` `instance-attribute` ¶

message `instance-attribute` ¶