Metadata-Version: 2.4
Name: retryxpy
Version: 0.1.0
Summary: Simple sync and async retry decorator with backoff and jitter
Author-email: Hossein Tajfirouz <hosseintajfirouz@gmail.com>
License: MIT
Project-URL: Homepage, https://github.com/ht21992/retryxpy
Project-URL: Repository, https://github.com/ht21992/retryxpy
Project-URL: Issues, https://github.com/ht21992/retryxpy/issues
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Framework :: AsyncIO
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Dynamic: license-file

# 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

```bash
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:

```bash
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)
```
