Calling circuit breaker by decorator

[lesnik512]

@NikitaKozlovtcev What do you think, will it be useful to have an opportunity using circuit breaker as a decorator?

From https://github.com/fabfuel/circuitbreaker:

from circuitbreaker import circuit

@circuit(failure_threshold=10, expected_exception=ConnectionError)
def external_call():
    ...

[NikitaKozlovtcev]

it looks like failure_threshold is not enough for circuit breaker. It should have ttl and some arguments about type of cash for storing falures by calls.

look how it can be

import asyncio
import logging
import typing
from functools import wraps

import httpx

from circuit_breaker_box import BaseCircuitBreaker, CircuitBreakerInMemory


logger = logging.getLogger(__name__)

MAX_RETRIES = 4
MAX_CACHE_SIZE = 256
CIRCUIT_BREAKER_MAX_FAILURE_COUNT = 2
RESET_TIMEOUT_IN_SECONDS = 10
SOME_HOST = "https://postman-echo.com/"

some_circuit_breaker = CircuitBreakerInMemory(
    reset_timeout_in_seconds=RESET_TIMEOUT_IN_SECONDS,
    max_failure_count=CIRCUIT_BREAKER_MAX_FAILURE_COUNT,
    max_cache_size=MAX_CACHE_SIZE,
)
T = typing.TypeVar("T")
P = typing.ParamSpec("P")


def circuit_breaker_decorator(circuit_breaker: BaseCircuitBreaker, expected_exception: type[Exception]) -> T:
    def decorator(func: typing.Callable[P, typing.Awaitable[T]]) -> T:
        @wraps(func)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> typing.Awaitable[T]:
            if not (await circuit_breaker.is_host_available(str(func))):
                await circuit_breaker.raise_host_unavailable_error(str(func))
            try:
                response = await func(*args, **kwargs)
            except expected_exception:
                await circuit_breaker.increment_failures_count(str(func))
            else:
                return response

        return wrapper

    return decorator


@circuit_breaker_decorator(circuit_breaker=some_circuit_breaker, expected_exception=httpx.HTTPStatusError)
async def external_call(url: str) -> tuple[int, dict]:
    client = httpx.AsyncClient(base_url=SOME_HOST, timeout=httpx.Timeout(10))
    response = await client.get(url)
    response.raise_for_status()
    return response.status_code, response.json()


async def main() -> None:
    logging.basicConfig(level=logging.INFO)

    for _ in range(MAX_RETRIES):
        result = await external_call("/status/500")
        logger.info("Final result: %s", result)


if __name__ == "__main__":
    asyncio.run(main())

[NikitaKozlovtcev]

do you want only one exception in expected_exception?

[lesnik512]

@NikitaKozlovtcev no, it's just an example