swarmauri_middleware_ratepolicy 0.11.0.dev1

Synchronous retry-policy middleware for Swarmauri with bounded attempts and exponential backoff.

Swarmauri Middleware Rate Policy

swarmauri_middleware_ratepolicy is the Swarmauri retry-policy middleware for wrapping request-like call flows with bounded retry attempts and exponential backoff. It provides a synchronous dispatch(request, call_next) surface that retries failing operations before finally re-raising the exception.

Why Use Swarmauri Middleware Rate Policy

• Retry flaky upstream operations with a consistent middleware surface.
• Bound retry counts while still applying exponential backoff.
• Add basic resilience to synchronous request-processing or job-execution pipelines.
• Combine with circuit breakers and logging to build layered fault handling.

FAQ

What parameters does this middleware expose?
max_retries and initial_wait.

How is the wait interval calculated?
The middleware sleeps for initial_wait * 2**attempts between retries.

Is the middleware async?
No. The current implementation is synchronous.

What happens after the retry limit is reached?
The last exception is re-raised.

Features

• Bounded retry loops over a synchronous call_next callable.
• Exponential backoff using initial_wait.
• Log messages for processing, retry attempts, and success.
• Swarmauri MiddlewareBase integration for reusable resilience flows.
• Supports Python 3.10, 3.11, 3.12, 3.13, and 3.14.

Installation

uv add swarmauri_middleware_ratepolicy

pip install swarmauri_middleware_ratepolicy

Usage

from swarmauri_middleware_ratepolicy import RetryPolicyMiddleware

retry = RetryPolicyMiddleware(max_retries=3, initial_wait=0.5)

class RequestEnvelope:
    def __init__(self, payload: str):
        self.payload = payload

request = RequestEnvelope("work-item-123")

def call_next(req):
    raise RuntimeError("Simulated upstream failure")

retry.dispatch(request, call_next)

Examples

Retry a failing operation

import logging
from swarmauri_middleware_ratepolicy import RetryPolicyMiddleware

logging.basicConfig(level=logging.INFO)

retry = RetryPolicyMiddleware(max_retries=3, initial_wait=0.25)

def unstable(_request):
    raise RuntimeError("Temporary failure")

retry.dispatch({"task": "sync"}, unstable)

Wrap an external API call

import httpx
from swarmauri_middleware_ratepolicy import RetryPolicyMiddleware

retry = RetryPolicyMiddleware(max_retries=4, initial_wait=0.25)

class RequestWrapper:
    def __init__(self, url: str):
        self.url = url

response = retry.dispatch(
    RequestWrapper("https://api.example.com/data"),
    lambda req: httpx.get(req.url, timeout=5),
)

print(response.status_code)

Register in a larger middleware stack

from swarmauri_middleware_ratepolicy import RetryPolicyMiddleware

middleware = RetryPolicyMiddleware(max_retries=2, initial_wait=1.0)
print(middleware.type)

Related Packages

• swarmauri_middleware_circuitbreaker
• swarmauri_state_clipboard
• swarmauri_middleware_ratepolicy

Swarmauri Foundations

• swarmauri
• swarmauri_core
• swarmauri_base
• swarmauri_standard

More Documentation

• Tenacity documentation

Best Practices

• Keep retry counts small for latency-sensitive user requests.
• Use circuit breakers when a dependency is consistently unhealthy instead of only increasing retries.
• Apply retries around idempotent operations whenever possible.
• Convert async flows deliberately if you need to wrap them with this synchronous middleware surface.

License

This project is licensed under the Apache-2.0 License.