retryxpy 0.1.0

Simple sync and async retry decorator with backoff and jitter

retryxpy

Simple sync and async retry decorator for Python.

Features

• Sync and Async Support: Works with both def and async def functions automatically.
• Multiple Backoff Strategies Supports:
  • fixed
  • linear
  • exponential
  • Jitter Support: Adds random delay to retries to prevent retry storms and thundering herd problems.
• Exception-Based Retries: Retry only for specific exception types.
• Result-Based Retries: Retry when a function returns an undesired result.
• Retry Hooks Execute custom logic during retry lifecycle:
  • before_retry
  • after_retry
  • on_retry
  • Retry State Object
• Hooks receive a RetryState object containing:
  • attempt number
  • max attempts
  • delay before next retry
  • exception (if any)
  • result (if any)
  • elapsed retry time
  • function name
  • Logger Integration: Integrates with Python’s logging module to automatically log retry events.
• Timeout-Aware Retries: Stop retrying once the total retry window exceeds a specified timeout.
• Configurable Retry Policy Control:
  • max_attempts
  • delay
  • max_delay
  • jitter
  • timeout
  • No External Dependencies
• Pure Python implementation
  • Type-Safe Design: Uses modern typing features and dataclasses.
  • Lightweight and Fast: Minimal overhead suitable for production systems.

Installation

pip install retryxpy

Sync example

from retryxpy import retry

@retry(max_attempts=5, delay=1, backoff="exponential")
def fetch_data():
    ...

Async example

from retryxpy import retry

@retry(max_attempts=5, delay=0.5, backoff="linear")
async def fetch_data():
    ...

Retry only certain exceptions
@retry(
    max_attempts=3,
    delay=1,
    exceptions=(TimeoutError, ConnectionError),
)
def call_api():
    ...

Hook example

def on_retry(state):
    print(
        f"Attempt {state.attempt}/{state.max_attempts} failed "
        f"for {state.function_name}: {state.exception}. "
        f"Retrying in {state.delay:.2f}s"
    )

@retry(max_attempts=4, delay=1, on_retry=on_retry)
def work():
    ...

---

How to run locally

Create venv and install:

python -m venv .venv
source .venv/bin/activate
python -m pip install -U pip pytest pytest-asyncio build
python -m pip install -e .
Run tests:
pytest
Build package:
python -m build

Example usage

Fixed backoff

from retryxpy import retry

@retry(max_attempts=3, delay=1, backoff="fixed")
def save_data():
    print("trying...")
    raise ValueError("temporary")

Linear backoff

from retryxpy import retry

@retry(max_attempts=4, delay=1, backoff="linear")
def process_job():
    ...

Delays:

• 1s
• 2s
• 3s

Exponential backoff

from retryxpy import retry

@retry(max_attempts=5, delay=0.5, backoff="exponential")
def call_service():
...

Delays:

• 0.5s
• 1s
• 2s
• 4s

Jitter

from retryxpy import retry

@retry(max_attempts=5, delay=1, jitter=0.5)
def call_service():
...

Adds a random extra delay up to 0.5 seconds.

Result based retry

from retryxpy import retry

@retry(
    max_attempts=4,
    delay=1,
    retry_if_result=lambda result: result is None,
)
def get_cached_value():
    return None

Logger integration

import logging
from retryxpy import retry

logger = logging.getLogger("myapp.retry")

@retry(
    max_attempts=3,
    delay=1,
    logger=logger,
)
def fetch_data():
    ...

Before / after hooks

def before_retry(state):
    print("About to retry", state.attempt)

def after_retry(state):
    print("Retried", state.attempt)

@retry(
    max_attempts=3,
    delay=1,
    before_retry=before_retry,
    after_retry=after_retry,
)
def work():
    ...

Timeout-aware retries

from retryxpy import retry

@retry(
    max_attempts=10,
    delay=1,
    timeout=5,
)
def call_service():
    ...

Real scenario examples

Example 1 — HTTP API retry

import random
from retryxpy import retry


@retry(max_attempts=5, delay=1, backoff="exponential", jitter=0.5)
def fetch_weather(city: str):
    print("Calling weather API...")

    # simulate unstable network
    if random.random() < 0.7:
        raise ConnectionError("API temporarily unavailable")

    return {"city": city, "temp": 21}


data = fetch_weather("Paris")
print("Result:", data)

Possible output:

Calling weather API...
Calling weather API...
Calling weather API...
Result: {'city': 'Paris', 'temp': 21}

⸻

Example 2 — Database retry

import random
from retryxpy import retry


@retry(
    max_attempts=4,
    delay=0.5,
    backoff="linear",
    exceptions=(TimeoutError,)
)
def save_order(order_id: int):
    print(f"Saving order {order_id}...")

    if random.random() < 0.6:
        raise TimeoutError("DB timeout")

    print("Order saved")

save_order(1001)

⸻

Example 3 — Logging retries with hook

from retryxpy import retry


def retry_logger(state):
    print(
        f"Retry {state.attempt}/{state.max_attempts} "
        f"for {state.function_name} after {state.delay:.2f}s "
        f"due to {state.exception}"
    )


@retry(
    max_attempts=3,
    delay=1,
    backoff="exponential",
    on_retry=retry_logger,
)
def unstable_job():
    raise RuntimeError("Temporary failure")


unstable_job()

Output:

Retry 1/3 for unstable_job after 1.00s due to Temporary failure
Retry 2/3 for unstable_job after 2.00s due to Temporary failure

⸻

Example 4 — Async API retry

import random
import asyncio
from retryxpy import retry


@retry(max_attempts=5, delay=0.5, backoff="exponential")
async def fetch_user(user_id: int):
    print("Fetching user...")

    if random.random() < 0.7:
        raise RuntimeError("Service overloaded")

    return {"id": user_id, "name": "Alice"}


async def main():
    user = await fetch_user(10)
    print(user)


asyncio.run(main())

⸻

Example 5 — Retry file operation

from retryxpy import retry


@retry(max_attempts=3, delay=1)
def read_file():
    print("Reading file...")
    raise OSError("File temporarily locked")


read_file()

⸻

Example 6 — Retry until service becomes ready

import random
from retryxpy import retry


service_ready = False


@retry(max_attempts=6, delay=1, backoff="exponential")
def wait_for_service():
    global service_ready

    print("Checking service...")

    if not service_ready:
        if random.random() < 0.8:
            raise RuntimeError("Service not ready")
        service_ready = True

    return "Service ready"


print(wait_for_service())

Example 7 - API client with logging, hooks, result-retry

import logging
import random
from retryxpy import retry

logger = logging.getLogger("weather")
logging.basicConfig(level=logging.WARNING)


def before_retry(state):
    print(f"Preparing retry {state.attempt} for {state.function_name}")


def after_retry(state):
    print(f"Retry finished, waiting {state.delay:.2f}s")


def on_retry(state):
    print("Retry triggered because:", state.exception or state.result)


@retry(
    max_attempts=5,
    delay=1,
    backoff="exponential",
    jitter=0.5,
    retry_if_result=lambda r: r == {},
    before_retry=before_retry,
    after_retry=after_retry,
    on_retry=on_retry,
    logger=logger,
)
def fetch_weather(city: str):

    print("Calling weather API...")

    # simulate API problems
    r = random.random()

    if r < 0.4:
        raise ConnectionError("API unavailable")

    if r < 0.7:
        return {}  # empty result → trigger retry

    return {"city": city, "temp": 21}


data = fetch_weather("Paris")
print("Result:", data)

Example 8 - Database job with timeout-aware retries

import random
from retryxpy import retry


def retry_monitor(state):
    print(
        f"[Retry {state.attempt}] "
        f"{state.function_name} failed with {state.exception}. "
        f"Elapsed: {state.elapsed:.2f}s"
    )


@retry(
    max_attempts=20,
    delay=0.5,
    backoff="linear",
    jitter=0.3,
    timeout=5,
    exceptions=(TimeoutError,),
    on_retry=retry_monitor,
)
def save_order(order_id: int):

    print("Saving order...")

    if random.random() < 0.8:
        raise TimeoutError("DB lock")

    return True


save_order(1001)