gentlify 1.6.2

Adaptive async rate limiting for Python, closed-loop feedback control for API concurrency.

gentlify

Adaptive async rate limiting for Python, closed-loop feedback control for API concurrency.

Zero dependencies. Asyncio-native. Fully typed.

Gentlify automatically adjusts concurrency and dispatch rate in response to failures, so your application backs off when an API is struggling and speeds up when it recovers — without manual tuning.

Installation

pip install gentlify

Requires Python 3.11+.

Quick Start

import asyncio
from gentlify import Throttle

throttle = Throttle(max_concurrency=5)

async def main():
    for item in range(20):
        async with throttle.acquire() as slot:
            await call_api(item)

asyncio.run(main())

If requests start failing, gentlify automatically halves concurrency, enters a cooling period, then gradually reaccelerates — all without any manual intervention.

Context Manager API

The primary API uses acquire() as an async context manager:

async with throttle.acquire() as slot:
    result = await call_api(item)
    slot.record_tokens(result.token_count)  # optional token tracking

On success, gentlify records the completion and checks whether to reaccelerate. On exception, it records the failure and may decelerate if the failure threshold is reached.

Decorator API

Wrap async functions directly:

@throttle.wrap
async def call_api(item):
    return await httpx.post("/api", json=item)

# Each call is automatically throttled
await call_api(my_item)

The decorator preserves the function signature and return value. Failures are recorded automatically.

Token Budget

Track and enforce token consumption within a rolling time window:

from gentlify import Throttle, TokenBudget

throttle = Throttle(
    max_concurrency=10,
    token_budget=TokenBudget(max_tokens=100_000, window_seconds=60.0),
)

async with throttle.acquire() as slot:
    result = await call_llm(prompt)
    slot.record_tokens(result.usage.total_tokens)

When the budget is exhausted, acquire() blocks until tokens expire from the rolling window.

Circuit Breaker

Automatically stop sending requests when an API is down:

from gentlify import Throttle, CircuitBreakerConfig

throttle = Throttle(
    max_concurrency=10,
    circuit_breaker=CircuitBreakerConfig(
        consecutive_failures=5,
        open_duration=30.0,
        half_open_max_calls=2,
    ),
)

After 5 consecutive failures the circuit opens, rejecting requests with CircuitOpenError for 30 seconds. It then enters half-open state, allowing 2 probe requests. If those succeed, the circuit closes; if they fail, it re-opens with a doubled delay (capped at 5x).

Retry

Automatically retry failed requests with configurable backoff — no need for a separate retry library:

from gentlify import Throttle, RetryConfig

throttle = Throttle(
    max_concurrency=5,
    retry=RetryConfig(
        max_attempts=3,
        backoff="exponential_jitter",
        base_delay=1.0,
        max_delay=60.0,
    ),
)

@throttle.wrap
async def call_api(item):
    return await httpx.post("/api", json=item)

Retries happen inside the throttled slot, so concurrency accounting stays correct. Only the final failure (after all retries are exhausted) triggers throttle deceleration — intermediate failures just trigger backoff sleep.

Backoff strategies:

• fixed — constant delay between retries
• exponential — delay doubles each attempt (base_delay × 2^attempt, capped at max_delay)
• exponential_jitter — exponential with random jitter (default, recommended)

You can also filter which exceptions are retryable:

retry = RetryConfig(
    max_attempts=3,
    retryable=lambda exc: isinstance(exc, (TimeoutError, RateLimitError)),
)

Non-retryable exceptions propagate immediately without further attempts.

Configuration

From code

throttle = Throttle(
    max_concurrency=10,
    initial_concurrency=3,
    min_dispatch_interval=0.2,
    failure_threshold=3,
    cooling_period=10.0,
    total_tasks=1000,
    on_progress=lambda snap: print(f"{snap.percentage:.0f}%"),
)

From a dictionary

throttle = Throttle.from_dict({
    "max_concurrency": 10,
    "token_budget": {"max_tokens": 50000, "window_seconds": 60.0},
})

From environment variables

# Set GENTLIFY_MAX_CONCURRENCY=10, GENTLIFY_MIN_DISPATCH_INTERVAL=0.5, etc.
throttle = Throttle.from_env()

# Or with a custom prefix:
throttle = Throttle.from_env(prefix="MYAPP")

Callbacks

State change events

def on_change(event):
    print(f"[{event.kind}] {event.data}")

throttle = Throttle(
    max_concurrency=10,
    on_state_change=on_change,
)
# Prints: [decelerated] {'concurrency': (10, 5), ...}
# Prints: [reaccelerated] {'concurrency': (5, 6), ...}

Progress milestones

throttle = Throttle(
    max_concurrency=10,
    total_tasks=100,
    on_progress=lambda snap: print(
        f"{snap.percentage:.0f}% done, ETA {snap.eta_seconds:.0f}s"
    ),
)

Graceful Shutdown

# Stop accepting new requests
throttle.close()

# Wait for in-flight requests to finish
await throttle.drain()

After close(), any new acquire() call raises ThrottleClosed. In-flight requests complete normally. drain() blocks until all in-flight requests finish.

Snapshot

Inspect the throttle's current state at any time:

snap = throttle.snapshot()
print(snap.concurrency)        # current concurrency limit
print(snap.dispatch_interval)  # current dispatch interval
print(snap.state)              # RUNNING, COOLING, CIRCUIT_OPEN, etc.
print(snap.tokens_remaining)   # remaining token budget (or None)
print(snap.eta_seconds)        # estimated time remaining (or None)

Types

All public types are re-exported from the top-level package:

Type	Description
Throttle	Main orchestrator
ThrottleConfig	Validated configuration dataclass
TokenBudget	Token budget configuration
CircuitBreakerConfig	Circuit breaker configuration
RetryConfig	Retry and backoff configuration
ThrottleSnapshot	Read-only state view
ThrottleState	Enum: RUNNING, COOLING, CIRCUIT_OPEN, CLOSED, DRAINING
ThrottleEvent	Structured event for state change callbacks
GentlifyError	Base exception
CircuitOpenError	Raised when circuit breaker is open
ThrottleClosed	Raised when throttle is closed

Development

pip install -e ".[dev]"
pytest
mypy --strict src/gentlify
ruff check src/ tests/

Releasing

1. Bump the version in src/gentlify/_version.py and pyproject.toml
2. Update CHANGELOG.md
3. Commit and push to main
4. Tag the release and push:

   git tag v<version>
   git push --tags
   

5. The GitHub Action builds and publishes to PyPI automatically via trusted publishing (OIDC)

License

Apache-2.0 — Copyright (c) 2026 Pointmatic