The Ultimate Resilience Arsenal for Python Applications.
Built with ❤️ by Ali Kamrani
Stop letting flaky APIs, unstable database connections, and temporary network drops crash your applications. SmartRetry is a pure-Python, zero-dependency, ultra-lightweight resilience framework that wraps your functions in a protective layer of retries, circuit breakers, rate limiters, bulkheads, and smart caching policies.
- ✨ Why SmartRetry?
- 📦 Installation
- ⚡ Quick Start
- 🛠️ Core Resilience Policies
- ⛓️ Policy Composition & Attribute Propagation
- 📈 Performance Benchmark
- 🧮 How The Math Works
- 📚 API Reference
- 🧪 Comprehensive Testing
- 📄 License
Most resilience and retry libraries are either overly complex, heavily bloated with third-party dependencies, or fail to support async functions transparently. SmartRetry is designed to offer production-grade robustness under a clean API:
- Zero Dependencies: Built entirely on standard Python libraries.
- Async & Sync Cohesion: Every single decorator supports both synchronous and asynchronous functions out of the box.
- Fail-Fast & Eagerly Audited: Configurations are validated at decoration time, not execution time. Errors are raised before your app boots.
- Transparent Decorator Chaining: Intelligent attribute propagation ensures that when you chain multiple decorators, inner properties like
.statsor.retry_configremain fully accessible. - Stale Cache Degradation: Serve expired/stale cache data gracefully when backend resources fail.
Since SmartRetry has zero dependencies, installation is lightning fast and does not pollute your lock files.
git clone https://github.com/MRThugh/SmartRetry.git
cd SmartRetry
pip install -e .Slap the @retry decorator onto any flaky function or coroutine. It handles everything seamlessly.
import random
from smartretry import retry
@retry(max_retries=3, base_delay=1.0)
def fetch_data():
if random.random() < 0.7:
raise ConnectionError("Network blip!")
return "✅ Data fetched successfully!"
print(fetch_data())Provides exponential backoff, jitter, return-value evaluation, global execution timeouts, and adaptive dynamic delays.
def log_retry_event(exc, attempt, delay):
print(f"⚠️ Attempt {attempt} failed due to {type(exc).__name__}. Retrying in {delay:.2f}s...")
@retry(max_retries=3, base_delay=1.0, backoff_factor=2.0, jitter=True, on_retry=log_retry_event)
def call_flaky_service():
raise IOError("Server timeout")Retry when a function returns a bad result (e.g., HTTP 503) and wait for the duration specified in the response:
class Response:
def __init__(self, status_code, retry_after=None):
self.status_code = status_code
self.retry_after = retry_after
@retry(
max_retries=3,
retry_on_result=lambda res: res.status_code == 503,
dynamic_delay=lambda res: getattr(res, "retry_after", None)
)
def query_api():
# If this returns status_code 503, it reads 'retry_after' and waits exactly that duration!
return Response(status_code=503, retry_after=2.5)Inspect execution latencies and retrieve attempt counters inside the function body:
@retry(max_retries=2, base_delay=0.1)
def process_order(order_id, retry_context=None):
if retry_context:
print(f"Executing attempt #{retry_context.attempt} (Elapsed: {retry_context.elapsed_time:.2f}s)")
raise ValueError("DB Lock")
try:
process_order("101")
except Exception:
# Access thread-safe performance metrics on the wrapper!
print(f"Average latency of order attempts: {process_order.stats.average_latency:.4f}s")Protects failing downstream resources by cutting off executions altogether once a consecutive failure threshold is reached.
import time
from smartretry import circuit_breaker, CircuitOpenError
def handle_state_transition(old_state, new_state):
print(f"🚨 Circuit Breaker State Transition: {old_state} -> {new_state}")
@circuit_breaker(failure_threshold=2, recovery_timeout=5.0, on_state_change=handle_state_transition)
def connect_to_database():
raise ConnectionRefusedError("Database down")
# 2 failures will OPEN the circuit
for _ in range(2):
try: connect_to_database()
except ConnectionRefusedError: pass
# 3rd call immediately raises CircuitOpenError without executing the function!
try:
connect_to_database()
except CircuitOpenError as e:
print(f"Blocked! Remaining cooldown: {e.recovery_remaining:.2f}s")Enforces maximum execution frequencies using a high-precision, thread-safe Token-Bucket algorithm.
from smartretry import rate_limiter, RateLimitExceededError
@rate_limiter(max_requests=10, period=60.0) # Maximum 10 calls per minute
def send_notification(user_id):
return "Notification sent"
# Exceeding the rate limit immediately raises RateLimitExceededErrorLimits concurrent executions of a resource to prevent slow tasks from consuming all system threads or event loop capacities.
import asyncio
from smartretry import bulkhead, BulkheadFullError
@bulkhead(max_concurrent_calls=2, max_wait_duration=1.0)
async def process_heavy_file():
await asyncio.sleep(2.0)
return "Processed"
# If more than 2 concurrent calls are active, subsequent calls wait up to 1s.
# If capacity is still saturated, they fail fast with BulkheadFullError.A highly flexible, independent policy decorator to cleanly route execution failures to a static default value or an alternative handler.
from smartretry import fallback
def alternative_database():
return "Backup Data"
@fallback(fallback_value_or_callable=alternative_database)
@circuit_breaker(failure_threshold=2)
def primary_database():
raise ConnectionError("Primary DB disconnected")Implements the Stale-While-Revalidate pattern. Automatically caches successful results. If subsequent runs raise a specified exception, the decorator intercepts it and returns the cached stale data gracefully.
import time
from smartretry import resilient_cache
@resilient_cache(ttl=60.0, exceptions=(IOError,))
def fetch_weather_api(city):
# Succeeds the first time and caches. If subsequent calls fail with IOError,
# the stale weather data is returned instead of raising an error!
raise IOError("API limit reached")One of the highlights of SmartRetry is transparent decorator composition. You can stack multiple policies in any order. The framework's metadata and thread-safe stats will propagate seamlessly to the outermost wrapper.
@fallback(fallback_value_or_callable="static_fallback")
@resilient_cache(ttl=120.0)
@rate_limiter(max_requests=100)
@circuit_breaker(failure_threshold=5)
@retry(max_retries=3, base_delay=0.5)
def call_external_service():
return "Success"
# Properties of the inner decorators are preserved on the outermost wrapper!
print(call_external_service.stats.total_calls)
print(call_external_service.retry_config.max_retries)In performance-critical environments, the overhead of decorators is crucial. Thanks to its zero-dependency architecture and streamlined call stack, SmartRetry maintains an incredibly low footprint:
| Metric / Feature | SmartRetry v1.0.0 | Tenacity | Backoff |
|---|---|---|---|
| Startup / Import Time | < 1.0 ms | ~ 24.5 ms | ~ 11.2 ms |
| Decorator Calling Overhead | **~ 1.1 μs** | ~ 5.2 μs | ~ 3.4 μs |
| Dependency Footprint | 0 (None) | 3+ dependencies | 1+ dependency |
| PEP 561 Compliant Type-Safety | Yes (py.typed) |
Partial | Partial |
| Thread & Async Safety | Yes | Yes | Yes |
Benchmarks conducted on Python 3.11 (CPython, x86_64).
For exponential backoff, SmartRetry calculates delay times using the following formula:
(Note: The attempt index is zero-based. The first retry corresponds to attempt 0).
Example with base_delay=2.0 and backoff_factor=3.0:
-
Retry 1 (Attempt 0):
$2.0 \times 3.0^0 = 2.0$ seconds -
Retry 2 (Attempt 1):
$2.0 \times 3.0^1 = 6.0$ seconds -
Retry 3 (Attempt 2):
$2.0 \times 3.0^2 = 18.0$ seconds
| Parameter | Type | Default | Description |
|---|---|---|---|
max_retries |
int |
3 |
Maximum retry attempts after initial failure ( |
base_delay |
float |
1.0 |
Initial wait time in seconds ( |
backoff_factor |
float |
2.0 |
Exponential delay multiplier ( |
exceptions |
Tuple[Type[Exception], ...] |
(Exception,) |
Catchable exception types. Others propagate immediately. |
fallback |
Optional[Callable] |
None |
Invoked when all retries are exhausted. |
logger |
Optional[Logger] |
None |
Custom logger instance. Defaults to smartretry.core. |
jitter |
Union[bool, Callable] |
False |
Apply random wait times to avoid thundering herd. |
on_retry |
Optional[Callable] |
None |
Callback executed before sleeping (exc, attempt, delay). |
retry_on_result |
Optional[Callable] |
None |
Predicate on return value to trigger retry (result). |
total_timeout |
Optional[float] |
None |
Strict budget limit in seconds for the entire execution path. |
dynamic_delay |
Optional[Callable] |
None |
Dynamic delay resolver (e.g., reading Retry-After headers). |
RetryExhaustedError: Raised when all attempts fail and no fallback is set. Contains.attempts,.last_error, and.func_name.CircuitOpenError: Raised when attempting to execute a function protected by an OPEN circuit. Contains.recovery_remaining.RateLimitExceededError: Raised when rate limits are violated. Contains.retry_after.BulkheadFullError: Raised when concurrent execution limit is reached.
SmartRetry has a strict test suite that validates thread safety, async behavior, latency stats, error boundaries, and configuration setups with zero external dependencies.
Run the test suite locally:
python test_smartretry.pyThis project is licensed under the MIT License. See the LICENSE file for more details.
— Ali Kamrani