feat(apis): circuit breaker and rate_limiter

Author: arthurbrenno

agentle/agents/apis/circuit_breaker.py

@@ -1,74 +1,69 @@
-"""Circuit breaker for resilient API calls."""
+"""Circuit breaker for resilient API calls.
+
+This module adapts the resilience module's circuit breaker implementations
+for use in the APIs module, maintaining backward compatibility.
+"""
 
 from __future__ import annotations
 
-import asyncio
-import time
 from collections.abc import Callable
 from typing import Any
 
 from agentle.agents.apis.circuit_breaker_error import CircuitBreakerError
-from agentle.agents.apis.circuit_breaker_state import CircuitBreakerState
 from agentle.agents.apis.request_config import RequestConfig
+from agentle.resilience.circuit_breaker.in_memory_circuit_breaker import (
+    InMemoryCircuitBreaker,
+)
 
 
 class CircuitBreaker:
-    """Circuit breaker implementation for resilient API calls."""
+    """
+    Circuit breaker implementation for resilient API calls.
+
+    This wraps the resilience module's InMemoryCircuitBreaker to provide
+    a simpler call-based API for endpoint usage.
+    """
 
     def __init__(self, config: RequestConfig):
         self.config = config
-        self.state = CircuitBreakerState.CLOSED
-        self.failure_count = 0
-        self.success_count = 0
-        self.last_failure_time: float | None = None
-        self._lock = asyncio.Lock()
+        self._circuit_id = "default"  # Single circuit per endpoint
+        # Initialize the underlying circuit breaker from resilience module
+        self._impl = InMemoryCircuitBreaker(
+            failure_threshold=config.circuit_breaker_failure_threshold,
+            recovery_timeout=config.circuit_breaker_recovery_timeout,
+            half_open_success_threshold=config.circuit_breaker_success_threshold,
+            enable_metrics=config.enable_metrics,
+        )
 
     async def call(self, func: Callable[[], Any]) -> Any:
-        """Execute function with circuit breaker protection."""
-        async with self._lock:
-            # Check if circuit is open
-            if self.state == CircuitBreakerState.OPEN:
-                # Check if we should transition to half-open
-                if self.last_failure_time:
-                    elapsed = time.time() - self.last_failure_time
-                    if elapsed >= self.config.circuit_breaker_recovery_timeout:
-                        self.state = CircuitBreakerState.HALF_OPEN
-                        self.success_count = 0
-                    else:
-                        raise CircuitBreakerError(
-                            f"Circuit breaker is OPEN. Retry after {self.config.circuit_breaker_recovery_timeout - elapsed:.1f}s"
-                        )
+        """
+        Execute function with circuit breaker protection.
+
+        Args:
+            func: Async function to execute
+
+        Returns:
+            Result of func call
+
+        Raises:
+            CircuitBreakerError: If circuit is open
+        """
+        # Check if circuit is open
+        if await self._impl.is_open(self._circuit_id):
+            # Get circuit state for more details
+            state = await self._impl.get_circuit_state(self._circuit_id)
+            next_retry_seconds = state.get("next_recovery_attempt_in_seconds", 0)
+
+            if next_retry_seconds > 0:
+                raise CircuitBreakerError(
+                    f"Circuit breaker is OPEN. Retry after {next_retry_seconds:.1f}s"
+                )
 
         # Execute the function
         try:
             result = await func()
-            await self._on_success()
+            await self._impl.record_success(self._circuit_id)
             return result
         except Exception:
-            await self._on_failure()
+            await self._impl.record_failure(self._circuit_id)
             raise
-
-    async def _on_success(self) -> None:
-        """Handle successful call."""
-        async with self._lock:
-            if self.state == CircuitBreakerState.HALF_OPEN:
-                self.success_count += 1
-                if self.success_count >= self.config.circuit_breaker_success_threshold:
-                    self.state = CircuitBreakerState.CLOSED
-                    self.failure_count = 0
-            elif self.state == CircuitBreakerState.CLOSED:
-                self.failure_count = 0
-
-    async def _on_failure(self) -> None:
-        """Handle failed call."""
-        async with self._lock:
-            self.failure_count += 1
-            self.last_failure_time = time.time()
-
-            if self.state == CircuitBreakerState.HALF_OPEN:
-                # Failure in half-open state reopens circuit
-                self.state = CircuitBreakerState.OPEN
-            elif self.state == CircuitBreakerState.CLOSED:
-                # Check if we've hit the failure threshold
-                if self.failure_count >= self.config.circuit_breaker_failure_threshold:
-                    self.state = CircuitBreakerState.OPEN

agentle/agents/apis/rate_limiter.py

@@ -1,39 +1,57 @@
-"""Rate limiter for API calls."""
+"""Rate limiter for API calls.
+
+This module adapts the resilience module's rate limiter implementations
+for use in the APIs module, maintaining backward compatibility.
+"""
 
 from __future__ import annotations
 
 import asyncio
-import time
 
 from agentle.agents.apis.request_config import RequestConfig
+from agentle.resilience.rate_limiting.rate_limit_config import RateLimitConfig
+from agentle.resilience.rate_limiting.in_memory_rate_limiter import (
+    InMemoryRateLimiter as ResilienceRateLimiter,
+)
 
 
 class RateLimiter:
-    """Rate limiter for API calls."""
+    """
+    Rate limiter for API calls.
+
+    This wraps the resilience module's InMemoryRateLimiter to provide
+    a simpler acquire-based API for endpoint usage.
+    """
 
     def __init__(self, config: RequestConfig):
         self.config = config
-        self.calls: list[float] = []
-        self._lock = asyncio.Lock()
+        self._identifier = "default"  # Single rate limit per endpoint
+
+        # Convert rate limit config to resilience module format
+        rate_limit_config: RateLimitConfig = {
+            "max_requests_per_minute": int(
+                config.rate_limit_calls * (60 / config.rate_limit_period)
+            )
+            if config.rate_limit_period <= 60
+            else config.rate_limit_calls,
+        }
+
+        # Initialize the underlying rate limiter from resilience module
+        self._impl = ResilienceRateLimiter(
+            default_config=rate_limit_config,
+            enable_metrics=config.enable_metrics,
+        )
 
     async def acquire(self) -> None:
-        """Acquire rate limit slot, waiting if necessary."""
-        async with self._lock:
-            now = time.time()
-
-            # Remove old calls outside the window
-            cutoff = now - self.config.rate_limit_period
-            self.calls = [t for t in self.calls if t > cutoff]
-
-            # Check if we're at the limit
-            if len(self.calls) >= self.config.rate_limit_calls:
-                # Calculate wait time
-                oldest_call = self.calls[0]
-                wait_time = self.config.rate_limit_period - (now - oldest_call)
-                if wait_time > 0:
-                    await asyncio.sleep(wait_time)
-                    # Recursively try again
-                    return await self.acquire()
-
-            # Record this call
-            self.calls.append(now)
+        """
+        Acquire rate limit slot, waiting if necessary.
+
+        This will block until a slot is available.
+        """
+        # Wait until we can proceed
+        while not await self._impl.can_proceed(self._identifier):
+            # Wait a short time before checking again
+            await asyncio.sleep(0.1)
+
+        # Record the request
+        await self._impl.record_request(self._identifier)