🌿 tranq

Calm, production-grade error handling & resilience for Python — decorator-based, zero boilerplate, battle-tested.

Quick Start • Features • API Reference • Examples • Comparison • Installation

🧘 Why tranq?

Writing repetitive try/except blocks clutters your code and buries business logic under layers of defensive programming. Retry loops, circuit breakers, rate limits, and timeout handling are cross-cutting concerns that should not pollute your core logic.

tranq gives you declarative error handling through decorators, context managers, and a comprehensive resilience toolkit — so you focus on what your code does, not how it recovers from failure.

# ❌ Without tranq: 30+ lines of boilerplate
import time, logging
for attempt in range(5):
    try:
        result = call_external_api()
        break
    except ConnectionError as e:
        if attempt == 4: raise
        logging.warning(f"Attempt {attempt+1} failed: {e}")
        time.sleep(2 ** attempt)

# ✅ With tranq: 2 lines
import tranq

@tranq.handle(on=ConnectionError, retry=4, delay=1.0, backoff=2.0)
def call_external_api():
    ...

Design Principles

⚡ Quick Start

Decorator (@handle)

import tranq

@tranq.handle(on=ValueError, retry=3, delay=0.5, backoff=2.0)
def risky_operation():
    """Retried up to 3 times on ValueError with exponential backoff."""
    ...

Async Decorator (@handle_async)

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

Circuit Breaker

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

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

Context Manager (Sync)

with tranq.retry(on=ValueError, retry=2, delay=0.1) as ctx:
    result = ctx.run(my_function, arg1, kwarg1=value)

Async Context Manager

async with tranq.retry_async(on=ConnectionError, retry=3) as ctx:
    result = await ctx.run(my_async_function, arg1)

Policy Composition DSL

from tranq import PolicyBuilder, wait, stop, retry_if

policy = (
    PolicyBuilder()
    .retry(max_attempts=3, wait=wait.exponential(0.5, max=10))
    .timeout(5)
    .circuit_breaker(tranq.SlidingWindowCircuitBreaker(failure_rate_threshold=0.5))
    .rate_limit(rate=10)
    .bulkhead(max_concurrent=5)
    .fallback(lambda: {"status": "cached"})
    .cache(ttl=60)
)

@policy
def call_service():
    ...

Rate Limiter

@tranq.rate_limit(rate=10, per=1.0, burst=20)
def api_call():
    """Limited to 10 calls/second with burst of 20."""
    ...

Bulkhead (Concurrency Limit)

@tranq.bulkhead(max_concurrent=5, timeout=10.0)
def limited_operation():
    """At most 5 concurrent executions."""
    ...

Retry Group (All-or-Nothing)

group = tranq.retry_group(step1, step2, step3, on=Exception, retry=2)
results = group.run()  # All steps retried together if any fails

📦 Installation

pip install tranq

Requires Python 3.9 or later. The library has zero runtime dependencies.

Optional Extras

pip install tranq[rich]         # Pretty terminal logging with colors
pip install tranq[sentry]       # Sentry error reporting
pip install tranq[slack]        # Slack webhook notifications
pip install tranq[prometheus]   # Prometheus metrics exposition
pip install tranq[all]          # All integrations
pip install tranq[dev]          # Development tools (pytest, black, isort, build, twine)

Verify installation:

python -m tranq
# Output: tranq v1.1.0 - Calm error handling with advanced resilience features.

🔥 Features

Layer 1 — Competitor Parity

🔁 Smart Retries

Exponential, linear, Fibonacci, constant, random backoff. Full/equal/decorrelated jitter. Custom callable. Min/max delay cap. Composable with +.

from tranq import wait, stop

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

⏱️ Timeouts

Hard execution limits for sync & async functions. Per-attempt and total-operation deadlines.

@tranq.handle(on=Exception, retry=2, timeout=5.0)
def slow_operation(): ...

🚦 Circuit Breakers

Count-based and sliding-window (failure-rate). Slow-call detection. State-change events. Breaker registry. Sync & async.

swc = tranq.SlidingWindowCircuitBreaker(
    window_size=100, failure_rate_threshold=0.5, minimum_calls=10)

@tranq.handle(on=Exception, circuit_breaker=swc)
def unreliable_service(): ...

🚦 Rate Limiting

Token bucket, leaky bucket, fixed window, sliding window, adaptive rate limiting.

@tranq.rate_limit(rate=10, per=1.0, burst=20)
def api_call(): ...

🚪 Bulkhead Isolation

Thread-based, async semaphore, queue-based, per-key isolation.

@tranq.bulkhead(max_concurrent=5, timeout=10.0)
def database_query(): ...

💰 Retry Budget

Prevents retry storms under load.

budget = tranq.RetryBudget(ttl=60, ratio=0.2, min_tokens=10)

@tranq.handle(on=Exception, retry=5, retry_budget=budget)
def protected_call(): ...

🏇 Hedged Requests

Race duplicate async requests, take the first success. Quorum mode. Percentile-based delay.

@tranq.hedged(hedge_delay=0.1, max_hedges=2)
async def fast_fetch(): ...

🧪 Conditional Retry

Retry on specific exceptions, result values, HTTP status codes, exception chains, or Retry-After headers.

from tranq import retry_if

@tranq.handle(
    on=Exception,
    retry=3,
    retry_if=retry_if.http_status((429, 503)) | retry_if.exception_type(ConnectionError),
)
def call_api(): ...

📝 Reporters

File (JSON lines), Log, Sentry, Slack, Prometheus. Pluggable custom reporters.

reporters = [
    tranq.FileReporter("errors.jsonl"),
    tranq.LogReporter(),
    tranq.SentryReporter(dsn="..."),
    tranq.SlackReporter(webhook_url="..."),
    tranq.PrometheusReporter(),
]

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

🎭 Mock Error Injection

Inject exceptions for testing with configurable probability and seed.

with tranq.mock_errors(ConnectionError, probability=0.8, seed=42):
    result = my_function()

💉 Dependency Injection

Inject dependencies into decorated functions.

@tranq.handle(inject={"logger": logging.getLogger("app")})
def do_work(logger=None):
    logger.info("Working...")

🌐 Global Policy

Set defaults once, override per function.

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

Layer 2 — What Makes tranq Different

🧠 Adaptive Resilience

Automatically tunes policy based on observed latency, error rate, and traffic.

ctrl = tranq.AdaptiveController(base_timeout=10.0, base_retry=3)
ctrl.observe(latency=2.0, success=False)
rec = ctrl.recommend()
# {'timeout': 2.5, 'retry': 1, 'observed_error_rate': 1.0, ...}

🎯 Policy Presets

Ready-made configs for common backends.

from tranq import presets

@tranq.resilient(**presets.http())
def fetch_users(): ...

@tranq.resilient(**presets.database())
def run_query(): ...

# Available: http, database, redis, kafka, queue, grpc, llm

🌐 HTTP Intelligence

Automatic retryable status detection and Retry-After parsing.

from tranq import is_retryable, parse_retry_after, recommended_wait

if is_retryable(exception):
    wait_seconds = recommended_wait(exception, default=1.0)

🔭 OpenTelemetry Native

Automatic spans and counters for resilience operations. Graceful no-op when not installed.

@tranq.telemetry(service="payment")
def process_payment(): ...

📡 Unified Event Bus

Subscribe to 14 resilience lifecycle events.

bus = tranq.EventBus()
bus.subscribe(tranq.events.CircuitOpened, lambda e: alert(f"CB opened: {e.data}"))
bus.subscribe("*", lambda e: log(e))  # wildcard

Events: RetryStarted, RetryCompleted, CircuitOpened, CircuitClosed, CircuitHalfOpened, TimeoutTriggered, RateLimitExceeded, BulkheadRejected, FallbackTriggered, CacheHit, CacheMiss, HedgeStarted, OperationSucceeded, OperationFailed.

🧬 Distributed Resilience

Pluggable state backends (in-memory, Redis) for cluster-wide rate limits, circuit breakers, and budgets.

from tranq import RedisBackend, DistributedRateLimiter

backend = RedisBackend(url="redis://localhost:6379/0")
limiter = DistributedRateLimiter(limit=100, window=1.0, backend=backend)

🕸️ Dependency-Aware Resilience

Track health of named dependencies and aggregate service health.

graph = tranq.DependencyGraph()
graph.register("api", dependencies=["postgres", "redis", "payment"])
graph.record("payment", success=False)
print(graph.failing_dependencies("api"))  # ["payment"]

🩺 Automatic Health Analysis

Full resilience diagnostics with recommended actions.

print(tranq.render_diagnostics())
# SERVICE HEALTH
# ────────────────────────────────
# API                  HEALTHY
# Payment              DEGRADED
# Overall: DEGRADED (calls=500, errors=12)
# Recommended actions:
#   - Enable jitter to avoid thundering herds
#   - A circuit breaker is open; investigate downstream

🧪 Chaos Testing Framework

Inject latency, errors, and timeouts for resilience testing.

with tranq.chaos(latency=0.3, error_rate=0.2, timeout_rate=0.1) as inj:
    for _ in range(100):
        inj.maybe_inject()
        call_service()

print(tranq.chaos_report())

🧩 Policy Composition DSL

Fluent builder composing all resilience features into a single decorator.

from tranq import PolicyBuilder, wait, stop, retry_if

policy = (
    PolicyBuilder()
    .retry(max_attempts=3, wait=wait.full_jitter(0.5, max=10),
           stop=stop.after_attempt(3) | stop.after_delay(30))
    .timeout(5)
    .circuit_breaker(tranq.SlidingWindowCircuitBreaker(...))
    .rate_limit(rate=10)
    .bulkhead(max_concurrent=5)
    .fallback(tranq.FallbackChain(cached_fallback, static_fallback))
    .cache(ttl=60)
    .observe(event_bus=bus)
)

@policy
def call_service(): ...

🤖 LLM Resilience

Rate-limit handling, Retry-After, provider/model fallback, automatic failover.

@tranq.llm(max_attempts=5, providers=[anthropic_client, local_model])
async def ask(prompt): ...

🏆 Auto Policy Detection

Detect operation type and recommend/apply a policy automatically.

@tranq.auto(verbose=True)  # Recommends only (safe by default)
async def fetch_orders(): ...

@tranq.auto(apply=True)    # Actually applies the recommended policy
async def fetch_users(): ...

📊 Advanced Metrics & Statistics

Counts, error-rate, min/max/avg, p50/p95/p99 latencies, summary tables, health reports.

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

print(tranq.summary_table())
print(tranq.overall_health())

🗄️ Resilience Cache

TTL, LRU eviction, stampede prevention (single-flight), stale-if-error, negative caching, cache metrics.

@tranq.cache(ttl=60, maxsize=256)
def expensive_computation(x): ...

🔄 Fallback Strategies

Static, callable, async, chain, cached (stale-while-failing), conditional.

chain = tranq.FallbackChain(cached_result, redis_fallback, static_default)

@PolicyBuilder().fallback(chain)
def get_data(): ...

📚 API Reference

Decorators

Context Managers

Retry Primitives

Circuit Breakers

Rate Limiters

Bulkheads

Other Components

Exceptions

Utilities

📁 Examples

The examples/ directory contains 38 complete, runnable scripts:

python examples/run_all.py       # Run all 38 examples
python examples/07_circuit_breaker.py  # Run a single example

🧪 Testing

The test suite contains 190+ tests covering all features across sync, async, and thread-safety:

pip install -e ".[dev]"
pytest tests/ -v

Test categories: circuit breakers (count + sliding window), all backoff strategies, jitter, conditional retry, error handlers, fallback, dependency injection, stateful retry, retry groups, reporters, metrics, profiling, mock errors, global policy, timeout, event hooks, rate limiters (5 algorithms), bulkhead, retry budget, hedged requests, cache, chaos testing, presets, diagnostics, event bus, wait/stop/retry_if composition, policy builder.

⚖️ Comparison

🗂️ Project Structure

tranq/
├── .github/workflows/
│   ├── ci.yml                    # Test matrix (5 Python × 3 OS)
│   └── release.yml              # Auto-publish to PyPI + GitHub Release
├── docs/source/
│   ├── conf.py
│   └── index.rst                # Full Sphinx documentation
├── examples/                     # 38 runnable examples
│   ├── 01_basic_decorator.py
│   ├── ...
│   ├── 38_auto_llm.py
│   └── run_all.py
├── src/tranq/
│   ├── __init__.py              # Public API (101 exports)
│   ├── __main__.py              # python -m tranq
│   ├── decorators.py            # @handle / @handle_async
│   ├── context.py               # retry() / retry_async()
│   ├── composition.py           # PolicyBuilder / ResiliencePolicy
│   ├── circuit_breaker.py       # Count-based CB + slow-call + registry
│   ├── async_circuit_breaker.py # Async count-based CB
│   ├── sliding_window.py        # Failure-rate CB (sync + async)
│   ├── rate_limiter.py          # 5 rate limiting algorithms
│   ├── bulkhead.py              # 4 bulkhead types
│   ├── retry_budget.py          # Retry storm prevention
│   ├── hedge.py                 # Hedged requests + quorum
│   ├── cache.py                 # TTL/LRU/stampede/stale-if-error
│   ├── fallback.py              # Chain/cached/conditional fallback
│   ├── wait.py                  # 10 composable wait strategies
│   ├── stop.py                  # Composable stop conditions
│   ├── retry_predicates.py      # Composable retry predicates
│   ├── timeout.py               # Deadline + per-attempt timeout
│   ├── events.py                # Unified event bus (14 events)
│   ├── adaptive.py              # Adaptive resilience controller
│   ├── presets.py               # http/db/redis/kafka/grpc/llm
│   ├── http_intel.py            # HTTP status + Retry-After
│   ├── diagnostics.py           # Health analysis + recommendations
│   ├── chaos.py                 # Chaos testing framework
│   ├── telemetry.py             # OpenTelemetry integration
│   ├── distributed.py           # Distributed state backends
│   ├── dependency.py            # Dependency health graph
│   ├── llm.py                   # LLM resilience + provider failover
│   ├── auto.py                  # Auto policy detection
│   ├── policies.py              # Policy dataclass + global policy
│   ├── exceptions.py            # Exception hierarchy
│   ├── metrics.py               # Metrics + percentiles
│   ├── statistics.py            # Summary tables + health
│   ├── profiling.py             # Function profiling
│   ├── reporters.py             # File/Log/Sentry/Slack/Prometheus
│   ├── mock.py                  # Error injection
│   ├── retry_group.py           # All-or-nothing groups
│   ├── utils.py                 # Backoff, jitter, logging
│   └── py.typed                 # PEP 561 marker
├── tests/                        # 190+ tests
├── CHANGELOG.md
├── CONTRIBUTING.md
├── LICENSE
├── pyproject.toml
├── pytest.ini
└── README.md

🏭 Production Best Practices

1. Always use jitter — prevents thundering herd on simultaneous retries.
2. Set max_delay — caps exponential backoff to prevent unbounded waits.
3. Use retry budgets for high-traffic services — prevents retry storms.
4. Combine circuit breaker + sliding window — tolerate sporadic failures.
5. Add observability — metrics + reporters + event bus for every critical path.
6. Use timeouts for external calls — never let them hang indefinitely.
7. Graceful degradation — always provide a fallback for user-facing operations.

🤝 Contributing

Contributions are welcome! Please see CONTRIBUTING.md.

# Setup
pip install -e ".[dev]"

# Test
pytest tests/ -v

# Format
black src/ tests/ && isort src/ tests/

📋 Changelog

See CHANGELOG.md for full release history.

Latest: v1.1.0 — Resilience Platform release with composable retry primitives, cache with stampede prevention, 5 rate limiting algorithms, 4 bulkhead types, policy composition DSL, adaptive resilience, presets, HTTP intelligence, OpenTelemetry, event bus, distributed state, dependency graph, chaos testing, LLM resilience, auto policy detection, and 190+ tests.

📄 License

MIT © RaptorVampire