Relinker

A retry library that shows you exactly what it is doing — and warns you when something looks risky.

from relinker import RetryPolicy

policy = (
    RetryPolicy()
    .attempts(5)
    .on(TimeoutError, ConnectionError)
    .exponential_delay(base=1, maximum=30)
    .jitter(maximum=0.5)
)

result = policy.run(fetch_data)

Five total attempts. Retries only on network errors. Exponential backoff capped at 30 s. Jitter prevents synchronized retries under concurrency.

Before this runs against a real service, ask Relinker what it thinks:

print(policy.explain())     # plain-language description
print(policy.preview(5))    # estimated timing per attempt
print(policy.doctor())      # flagged risks, if any

---

Install

pip install relinker

Requires Python 3.10+. No runtime dependencies.

For local development:

git clone https://github.com/igors93/relinker.git
cd relinker
python -m venv .venv && source .venv/bin/activate
python -m pip install -e ".[dev]"

---

What it does

Describe the policy, not the loop. Relinker separates what should retry from how it runs. Policies are immutable objects — compose them, share them, inspect them.

Guidance built in. warnings() and doctor() flag risky configurations like infinite retry without delay or retrying all exceptions. You keep full control; Relinker just points things out.

Full visibility. RetryResult records every attempt, timing, and error type. simulate() and timeline() estimate behaviour before production. Structured logging excludes exception messages by default to avoid leaking sensitive data.

Sync and async, same API. Decorate a regular function or a coroutine function — the same policy works for both.

Zero required dependencies. The core package has no runtime requirements. HTTP helpers, presets, and testing utilities are included.

---

Features at a glance

Category	What's included
Entry points	@retry decorator · fluent RetryPolicy builder · presets (network, database, fast, …)
Stop strategies	by attempt count · by elapsed time · forever · composable AND / OR
Retry conditions	by exception type · by returned value · custom callback · TryAgain signal
Delays	fixed · linear · exponential · random · chain · state-aware · jitter · custom
HTTP helpers	Retry-After support · status-code conditions · http_retry_policy()
Shared capacity	RetryBudget — process-local, per-key rolling window
Results	RetryResult · attempt history · per-function statistics
Observability	structured logging · event hooks · debug()
Guidance	warnings() · doctor() · explain() · simulate() · timeline() · preview()
Execution	sync run() · async run_async() · sync/async context managers
Testing	for_testing() · custom sleep injection · sleep capture

---

Stability

Relinker 1.0 provides a stable public API for Python 3.10 through 3.13. Compatibility guarantees cover the documented exports and behaviors described in the compatibility policy. Release history lives in CHANGELOG.md.

See the migration guide when upgrading from an earlier version.

---

Quick Start

The smallest useful retry

from relinker import retry

@retry(attempts=3, delay=1, on=(TimeoutError,))
def fetch_data() -> str:
    return call_external_service()

If fetch_data() raises TimeoutError, Relinker tries again up to 3 times and waits 1 second between attempts.

Use a preset

from relinker import network

@network()
def call_api() -> dict:
    return client.get("/users/1")

Presets are regular policies. You can keep customizing them:

policy = network().attempts(8).fallback_value({"status": "offline"})

Share a retry budget

A retry budget limits additional attempts across executions that share the same budget object and key. The original attempt is never counted.

from relinker import RetryBudget, RetryPolicy

budget = RetryBudget(max_retries=20, per=60)

policy = (
    RetryPolicy()
    .attempts(5)
    .on(TimeoutError)
    .exponential_delay(base=1, maximum=30)
    .with_retry_budget(budget, key="external-api")
)

RetryBudget is in-memory and process-local. Separate processes do not share capacity. Normal policy delays and max_time() continue to apply. See Retry budgets for the complete behavior and scope.

Use the full builder

from relinker import RetryPolicy

policy = (
    RetryPolicy()
    .attempts(5)
    .on(TimeoutError, ConnectionError)
    .exponential_delay(base=1, maximum=30)
    .jitter(maximum=0.5)
    .fallback_value({"status": "unavailable"})
)

result = policy.run(fetch_data)

---

Guidance

Relinker does not try to control your application. It lets you make your own choices, but it helps you notice risky retry policies.

from relinker import RetryPolicy

policy = RetryPolicy().forever().on(Exception).no_delay()

print(policy.doctor().describe())

Example output:

Relinker policy health

Risk level: risky

Warnings:
- forever: This policy can retry forever.
- no_delay: This policy has no delay between attempts.
- tight_loop_risk: This policy can retry forever without sleeping.
- broad_exception: This policy retries all Exception subclasses.

Use explain() when you want to understand a policy in plain language:

print(policy.explain())

Use preview() when you want to estimate timing before running real code:

print(policy.preview(attempts=5))

---

HTTP retry

Relinker includes dependency-free HTTP helpers. They work with any response object that exposes .status_code or a dictionary with a "status_code" key.

from relinker import http_retry_policy

policy = http_retry_policy(
    attempts=5,
    statuses={429, 500, 502, 503, 504},
    respect_retry_after=True,
)

For lower-level control:

from relinker import RetryPolicy, retry_after_delay, retry_if_status

policy = (
    RetryPolicy()
    .attempts(5)
    .retry_if_result(retry_if_status({429, 500, 502, 503, 504}))
    .stateful_delay(retry_after_delay(default=1.0, maximum=60.0))
)

This is useful for APIs that return 429 Too Many Requests with a Retry-After header.

---

Observability

Human-readable logging

import logging
from relinker import RetryPolicy

logging.basicConfig(level=logging.INFO)

policy = RetryPolicy().attempts(3).on(TimeoutError).with_logging(level=logging.INFO)

Structured logging

policy = RetryPolicy().attempts(3).on(TimeoutError).with_structured_logging()

Structured logs exclude error messages by default because exception messages can contain tokens, URLs, payload fragments, or user data.

Events

from relinker import RetryEvent, RetryPolicy

def on_retry(event: RetryEvent) -> None:
    print(f"retrying after attempt {event.attempt_number}, delay={event.delay}")

policy = RetryPolicy().attempts(3).on(TimeoutError).on_retry(on_retry)

---

Results and statistics

Return a RetryResult when you want full visibility:

result = RetryPolicy().attempts(3).return_result().run(fetch_data)

print(result.summary())
print(result.story())

Decorated functions also receive retry statistics:

from relinker import network

@network()
def fetch_user() -> dict:
    return {"id": 1}

fetch_user()

print(fetch_user.retry_stats.to_dict())

---

Examples

Run examples from the project root:

python -m examples.basic_retry
python -m examples.retry_with_policy
python -m examples.retry_policy_doctor
python -m examples.retry_preview_and_explain
python -m examples.retry_http_retry_after
python -m examples.retry_structured_logging

See examples/README.md for the full list.

---

Documentation

Getting started	Install and write your first policy
Choosing a policy	Decision guide by situation
Feature map	Quick lookup: need → API
When not to retry	Idempotency, generators, permanent failures
Common mistakes	Risky patterns with safer alternatives
Troubleshooting	Symptom-by-symptom diagnosis
Production checklist	Review before deploying
Retry lifecycle	How one execution flows
Retry budgets	Shared capacity explained
HTTP retry	Status codes and Retry-After
Testing	Keep tests fast and deterministic
API reference	Full method and export reference
Compatibility policy	Stability guarantees

Full index: docs/README.md

---

Contributing

Bug reports, questions, and pull requests are welcome.

• Read CONTRIBUTING.md for local setup, code principles, and the pull request process.
• Open an issue to report a bug or propose a feature before writing code.
• Keep changes small and focused — one behaviour change per pull request.
• Every bug fix needs a regression test. Coverage must not decrease.

---

Security

Relinker validates all numeric inputs at construction time and caps Retry-After header values to prevent unexpectedly long sleeps.

To report a vulnerability, open a private security advisory on GitHub. Do not publish sensitive details publicly before the issue is reviewed.

See SECURITY.md for the full security policy.

---

License

MIT.