🔁 retrykit

A clean, simple retry decorator for Python — sync, async, backoff, jitter, and custom exceptions.

---

🤔 Why retrykit?

Network calls fail. APIs go down. Databases time out.

Instead of writing manual retry loops every time, just add @retry and it handles everything for you.

# Without retrykit — ugly, repetitive
for attempt in range(3):
    try:
        result = requests.get(url)
        break
    except Exception:
        if attempt == 2:
            raise
        time.sleep(1)

# With retrykit — clean and simple
@retry(times=3, delay=1)
def fetch():
    return requests.get(url)

---

📦 Installation

pip install retrykit

---

🚀 Quick Start

from retrykit import retry

@retry(times=3, delay=1)
def fetch_data():
    return requests.get("https://api.example.com").json()

That's it. If fetch_data() fails, it retries up to 3 times, waiting 1 second between attempts.

---

⚙️ All Options

@retry(
    times=3,           # Total attempts (default: 3)
    delay=1.0,         # Seconds to wait between retries (default: 1.0)
    backoff=1.0,       # Multiply delay by this after each retry (default: 1.0)
    jitter=False,      # Add small random delay to avoid thundering herd (default: False)
    on=None,           # List of exceptions to retry on (default: any Exception)
    on_retry=None,     # Callback called before each retry
)

---

📖 Examples

Basic Retry

from retrykit import retry

@retry(times=3, delay=2)
def connect_to_server():
    return requests.get("https://api.example.com")

---

Exponential Backoff

Delays grow: 1s → 2s → 4s → 8s

@retry(times=4, delay=1, backoff=2)
def fetch_with_backoff():
    return requests.get("https://api.example.com")

---

Retry Only on Specific Exceptions

@retry(times=3, delay=1, on=[ConnectionError, TimeoutError])
def connect_db():
    return db.connect()

# TypeError or ValueError won't be retried — they raise immediately

---

Jitter (Avoid Thundering Herd)

Adds a small random delay so multiple clients don't hammer a server at the same time.

@retry(times=3, delay=2, jitter=True)
def api_call():
    return requests.post("https://api.example.com/data", json=payload)

---

on_retry Callback

Run custom logic before each retry (logging, alerting, etc.)

def log_retry(attempt, exception, next_delay):
    print(f"Attempt {attempt} failed: {exception}. Retrying in {next_delay}s...")

@retry(times=3, delay=1, on_retry=log_retry)
def risky_operation():
    return do_something_risky()

---

Async Support

Works with async def functions automatically — no extra config needed.

@retry(times=3, delay=1)
async def async_fetch():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com") as resp:
            return await resp.json()

---

Real-World Example

import requests
from retrykit import retry

def log_retry(attempt, exc, delay):
    print(f"[Warning] Attempt {attempt} failed: {exc}. Retrying in {delay}s...")

@retry(
    times=5,
    delay=1,
    backoff=2,
    jitter=True,
    on=[requests.ConnectionError, requests.Timeout],
    on_retry=log_retry,
)
def fetch_weather(city: str):
    response = requests.get(
        f"https://api.openweathermap.org/data/2.5/weather?q={city}",
        timeout=5
    )
    response.raise_for_status()
    return response.json()

data = fetch_weather("Chennai")

---

🆚 retrykit vs tenacity

Feature	retrykit	tenacity
Simple decorator API	✅ Clean	⚠️ Verbose
Async support	✅ Auto-detected	✅ Manual
Exponential backoff	✅ backoff=2	✅ wait_exponential()
Jitter	✅ jitter=True	✅ wait_random()
Zero dependencies	✅ Pure stdlib	✅
Learning curve	✅ Beginner-friendly	⚠️ Complex

---

🧪 Running Tests

pip install pytest
pytest tests/ -v

---

📄 License

MIT — free to use in personal and commercial projects.

---

🤝 Contributing

Pull requests are welcome! Please open an issue first to discuss changes.

1. Fork the repo
2. Create a branch: git checkout -b feature/my-feature
3. Make your changes + add tests
4. Open a Pull Request