🌿 tranq

Calm error handling for Python – decorator-based, zero boilerplate.

🧘 Why tranq?

Writing repetitive try/except blocks clutters your code. tranq gives you declarative error handling with decorators, context managers, and smart retry strategies, so you can focus on business logic.

📦 Installation

pip install tranq

Requires Python 3.9 or later.

⚡ Quick Start

Decorator (@handle)

import tranq

@tranq.handle(on=ValueError, retry=3, delay=0.5, backoff=2.0)
def risky():
    ...

Async (@handle_async)

@tranq.handle_async(on=ConnectionError, retry=2, fallback=lambda: "offline")
async def fetch_data():
    ...

Circuit Breaker

cb = tranq.CircuitBreaker(failure_threshold=5, timeout=60)

@tranq.handle(circuit_breaker=cb)
def call_unstable_service():
    ...

Context Manager

with tranq.retry(on=ValueError, retry=2) as ctx:
    result = ctx.run(my_function, arg1, arg2)

Retry Group (all-or-nothing)

group = tranq.retry_group(step1, step2, step3, on=Exception, retry=1)
results = group.run()

🔍 Features in Depth

1. Retry with Backoff

Choose from exponential, linear, or Fibonacci backoff. Add jitter to avoid thundering herds.

@tranq.handle(
    on=TimeoutError,
    retry=5,
    delay=0.1,
    backoff=2.0,
    backoff_strategy="exponential",  # "linear", "fibonacci", or custom callable
    max_delay=10.0,
    jitter=True,
)
def fetch():
    ...

2. Conditional Retry

• retry_if – retry only when the exception matches a condition.
• retry_on_result – retry if the result is unacceptable (e.g., None).

@tranq.handle(
    on=requests.RequestException,
    retry_if=lambda e: e.response.status_code == 429,  # rate-limit
    retry=3,
)
def call_api():
    ...

3. Global Policy

Set defaults for your entire application once.

tranq.set_global_policy(tranq.Policy(
    retry=3,
    delay=0.5,
    backoff=2.0,
    reraise=False,
))

# All @handle calls now inherit these defaults
@tranq.handle(on=ValueError)
def my_func():
    ...

4. Reporters (JSON file, Sentry, Slack)

from tranq import FileReporter, SentryReporter, SlackReporter

reporters = [
    FileReporter("/var/log/tranq_errors.json"),  # Writes JSON lines
    SentryReporter(dsn="..."),
    SlackReporter(webhook_url="..."),
]

@tranq.handle(on=Exception, reporters=reporters)
def critical_task():
    ...

5. Metrics & Profiling

@tranq.handle(metrics=True, metric_prefix="myapp")
def expensive_op():
    ...

from tranq import get_metrics, profile, get_profile

@profile
def heavy_computation():
    ...

print(get_metrics())
print(get_profile("heavy_computation"))

🚀 Advanced Example

Combining multiple features for a robust API call:

cb = tranq.CircuitBreaker(failure_threshold=3, timeout=60)

@tranq.handle(
    on=requests.RequestException,
    retry=5,
    backoff_strategy="fibonacci",
    max_delay=30,
    jitter=True,
    retry_if=lambda e: e.response.status_code in (429, 503),
    circuit_breaker=cb,
    metrics=True,
    metric_prefix="api",
    reporters=[tranq.FileReporter("api_errors.json")],
    fallback=lambda: {"status": "fallback"},
)
def fetch_from_external_api():
    ...

📚 API Reference

• Decorators: handle(...), handle_async(...)
• Context Manager: retry(...)
• Retry Groups: retry_group(...), async_retry_group(...)
• Circuit Breakers: CircuitBreaker(...), AsyncCircuitBreaker(...)
• Policies: Policy, set_global_policy(), get_global_policy()
• Reporters: Reporter, FileReporter, SentryReporter, SlackReporter
• Utilities: get_metrics(), reset_metrics(), profile(), get_profile(), mock_errors()
• Exceptions: TranqError, RetryExhaustedError, CircuitBreakerError, ResultNotAcceptedError, RetryGroupError

📖 Full documentation, 20 runnable examples, and 120+ tests are available on GitHub.

📄 License

MIT © RaptorVampire

Happy error handling! 🧘