added retry method in base circuit breaker (#1)

* added retry method in base circuit breaker

* PR fixes

* split retry function from circuit breaker

* split retry function from circuit breaker

* RP fixes

Author: NikitaKozlovtcev

README.md

@@ -1,6 +1,6 @@
 # Python Circuit Breaker Box
 
-A Python implementation of the Circuit Breaker pattern with Redis and in-memory storage support.
+A Python implementation of the Circuit Breaker pattern.
 
 ## Features
 
@@ -10,8 +10,8 @@ A Python implementation of the Circuit Breaker pattern with Redis and in-memory
 - [![Python](https://img.shields.io/badge/Python-3776AB?style=for-the-badge&logo=python&logoColor=FFD43B)](https://python.org) 3.10-3.13 support.
 - ⚡ Asynchronous API
 - 🔧 Configurable parameters
+- 🔄 Retries by [tenacity](https://tenacity.readthedocs.io/en/latest/)
 - 🛠️ FastAPI integration through custom exceptions
-- 🔄 Exponential backoff by [tenacity](https://tenacity.readthedocs.io/en/latest/) with jitter for retries
 
 ## Installation
 ```bash

circuit_breaker_box/__init__.py

@@ -1,13 +1,20 @@
 from circuit_breaker_box.circuit_breaker_base import BaseCircuitBreaker
 from circuit_breaker_box.circuit_breaker_in_memory import CircuitBreakerInMemory
 from circuit_breaker_box.circuit_breaker_redis import CircuitBreakerRedis
+from circuit_breaker_box.common_types import ResponseType
 from circuit_breaker_box.errors import BaseCircuitBreakerError, HostUnavailableError
+from circuit_breaker_box.retryer_base import BaseRetrier
+from circuit_breaker_box.retryers import Retrier, RetrierCircuitBreaker
 
 
 __all__ = [
     "BaseCircuitBreaker",
     "BaseCircuitBreakerError",
+    "BaseRetrier",
     "CircuitBreakerInMemory",
     "CircuitBreakerRedis",
     "HostUnavailableError",
+    "ResponseType",
+    "Retrier",
+    "RetrierCircuitBreaker",
 ]

circuit_breaker_box/circuit_breaker_base.py

@@ -4,7 +4,7 @@
 
 
 @dataclasses.dataclass(kw_only=True, slots=True)
-class BaseCircuitBreaker:
+class BaseCircuitBreaker(abc.ABC):
     reset_timeout_in_seconds: int
     max_failure_count: int
 

circuit_breaker_box/circuit_breaker_in_memory.py

@@ -10,7 +10,7 @@
 logger = logging.getLogger(__name__)
 
 
-@dataclasses.dataclass(kw_only=True, slots=True)
+@dataclasses.dataclass(kw_only=True)
 class CircuitBreakerInMemory(BaseCircuitBreaker):
     max_cache_size: int
     cache_hosts_with_errors: TTLCache[typing.Any, typing.Any] = dataclasses.field(init=False)

circuit_breaker_box/circuit_breaker_redis.py

@@ -17,7 +17,7 @@ def _log_attempt(retry_state: tenacity.RetryCallState) -> None:
     logger.info("Attempt redis_reconnect: %s", retry_state)
 
 
-@dataclasses.dataclass(kw_only=True, slots=True)
+@dataclasses.dataclass(kw_only=True)
 class CircuitBreakerRedis(BaseCircuitBreaker):
     redis_connection: "aioredis.Redis[str]"
 

circuit_breaker_box/common_types.py

@@ -0,0 +1,4 @@
+import typing
+
+
+ResponseType = typing.TypeVar("ResponseType")

circuit_breaker_box/retryer_base.py

@@ -0,0 +1,44 @@
+import abc
+import dataclasses
+import logging
+import typing
+
+import tenacity
+
+from circuit_breaker_box import ResponseType
+
+
+logger = logging.getLogger(__name__)
+
+P = typing.ParamSpec("P")
+
+
+@dataclasses.dataclass(kw_only=True)
+class BaseRetrier(abc.ABC, typing.Generic[ResponseType]):
+    max_retries: int
+    reraise: bool = True
+    exceptions_to_retry: tuple[type[Exception]]
+    stop: tenacity.stop.stop_after_attempt = dataclasses.field(init=False)
+    wait_strategy: tenacity.wait.wait_exponential_jitter = dataclasses.field(init=False)
+    retry_cause: tenacity.retry_if_exception_type = dataclasses.field(init=False)
+
+    def __post_init__(self) -> None:
+        self.stop = tenacity.stop_after_attempt(self.max_retries)
+        self.wait_strategy = tenacity.wait_exponential_jitter()
+        self.retry_cause = tenacity.retry_if_exception_type(self.exceptions_to_retry)
+
+    @abc.abstractmethod
+    async def retry(
+        self,
+        coroutine: typing.Callable[P, typing.Awaitable[ResponseType]],
+        *args: P.args,
+        **kwargs: P.kwargs,
+    ) -> ResponseType: ...
+
+    @staticmethod
+    def _log_attempts(retry_state: tenacity.RetryCallState) -> None:
+        logger.info(
+            "Attempt: attempt_number: %s, outcome_timestamp: %s",
+            retry_state.attempt_number,
+            retry_state.outcome_timestamp,
+        )

circuit_breaker_box/retryers.py

@@ -0,0 +1,62 @@
+import dataclasses
+import logging
+import typing
+
+import tenacity
+
+from circuit_breaker_box import BaseCircuitBreaker, BaseRetrier, ResponseType
+
+
+logger = logging.getLogger(__name__)
+
+P = typing.ParamSpec("P")
+
+
+@dataclasses.dataclass(kw_only=True)
+class Retrier(BaseRetrier[ResponseType]):
+    async def retry(  # type: ignore[return]
+        self,
+        coroutine: typing.Callable[P, typing.Awaitable[ResponseType]],
+        *args: P.args,
+        **kwargs: P.kwargs,
+    ) -> ResponseType:
+        for attempt in tenacity.Retrying(  # noqa: RET503
+            stop=self.stop,
+            wait=self.wait_strategy,
+            retry=self.retry_cause,
+            reraise=self.reraise,
+            before=self._log_attempts,
+        ):
+            with attempt:
+                return await coroutine(*args, **kwargs)
+
+
+@dataclasses.dataclass(kw_only=True)
+class RetrierCircuitBreaker(BaseRetrier[ResponseType]):
+    circuit_breaker: BaseCircuitBreaker
+
+    async def retry(  # type: ignore[return]
+        self,
+        coroutine: typing.Callable[P, typing.Awaitable[ResponseType]],
+        *args: P.args,
+        **kwargs: P.kwargs,
+    ) -> ResponseType:
+        if not (host := str(kwargs.get("host", ""))):
+            msg = "'host' argument should be defined"
+            raise ValueError(msg)
+
+        for attempt in tenacity.Retrying(  # noqa: RET503
+            stop=self.stop,
+            wait=self.wait_strategy,
+            retry=self.retry_cause,
+            reraise=self.reraise,
+            before=self._log_attempts,
+        ):
+            with attempt:
+                if not await self.circuit_breaker.is_host_available(host):
+                    await self.circuit_breaker.raise_host_unavailable_error(host)
+
+                if attempt.retry_state.attempt_number > 1:
+                    await self.circuit_breaker.increment_failures_count(host)
+
+                return await coroutine(*args, **kwargs)

examples/example_circuit_breaker.py

@@ -1,35 +1,27 @@
 import asyncio
 import logging
-import typing
-
-import fastapi
 
 from circuit_breaker_box import CircuitBreakerInMemory
 
 
-HTTP_MAX_TRIES = 4
+MAX_RETRIES = 4
 MAX_CACHE_SIZE = 256
 CIRCUIT_BREAKER_MAX_FAILURE_COUNT = 1
 RESET_TIMEOUT_IN_SECONDS = 10
 SOME_HOST = "http://example.com/"
 
 
-class CustomCircuitBreakerInMemory(CircuitBreakerInMemory):
-    async def raise_host_unavailable_error(self, host: str) -> typing.NoReturn:
-        raise fastapi.HTTPException(status_code=500, detail=f"Host: {host} is unavailable")
-
-
 async def main() -> None:
     logging.basicConfig(level=logging.DEBUG)
-    circuit_breaker = CustomCircuitBreakerInMemory(
+    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,
     )
 
     assert await circuit_breaker.is_host_available(host=SOME_HOST)
 
-    for _ in range(HTTP_MAX_TRIES):
+    for _ in range(MAX_RETRIES):
         await circuit_breaker.increment_failures_count(host=SOME_HOST)
 
     assert await circuit_breaker.is_host_available(host=SOME_HOST) is False

examples/example_retry.py

@@ -0,0 +1,31 @@
+import asyncio
+import logging
+
+import httpx
+
+from circuit_breaker_box.retryers import Retrier
+
+
+MAX_RETRIES = 4
+MAX_CACHE_SIZE = 256
+CIRCUIT_BREAKER_MAX_FAILURE_COUNT = 3
+RESET_TIMEOUT_IN_SECONDS = 10
+SOME_HOST = "http://example.com/"
+
+
+async def main() -> None:
+    logging.basicConfig(level=logging.DEBUG)
+    retryer = Retrier[httpx.Response](
+        max_retries=MAX_RETRIES,
+        exceptions_to_retry=(ZeroDivisionError,),
+    )
+    example_request = httpx.Request("GET", httpx.URL("http://example.com"))
+
+    async def foo(request: httpx.Request, host: str) -> httpx.Response:  # noqa: ARG001
+        raise ZeroDivisionError
+
+    await retryer.retry(coroutine=foo, request=example_request, host=example_request.url.host)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())