Getting Started

Getting Started

🌐 throttlekit.in · 🧪 runnable examples — one self-contained script per axis (sync + async service door, direct Redis, cost via debit, the admit lifecycle, a FastAPI app).

Install

pip install throttlekit-py            # the gRPC ServiceBackend
pip install "throttlekit-py[redis]"   # + a redis client for the direct RedisBackend

Installed as throttlekit-py, imported as throttlekit. Requires Python ≥ 3.10. The wheel pulls grpcio + protobuf (the ServiceBackend stubs); the [redis] extra adds a redis client for the direct door.

The service door — ServiceBackend

Point it at a running throttlekit-server. The core (inside the service) computes every decision.

ServiceBackend is store-agnostic — the server it points at can be backed by in-process memory, Redis, Postgres, or DynamoDB (chosen with --store on the server); you send the same requests and the decision is the core's regardless. (Deno KV and Cloudflare are edge-runtime stores, reachable only inside those runtimes — not through the service door.) The core's Polyglot & Python guide has the full "all stores from Python" recipe (a Docker-compose stack per backend).

from throttlekit import ServiceBackend

with ServiceBackend("localhost:50051") as rl:
    d = rl.check("api", api_key)        # consume one unit against the "api" policy
    if not d.allowed:
        retry_after_ms = d.retry_after_ms
        ...                             # 429

For TLS/mTLS pass channel credentials; the default is an insecure channel (loopback/dev only):

import grpc
creds = grpc.ssl_channel_credentials(root_certificates=open("ca.pem", "rb").read())
rl = ServiceBackend("limiter.internal:50051", credentials=creds)

Tip — see your decisions live. Start the server with --tui for ThrottleKit Lens, a built-in terminal dashboard (throttlekit-server --config x.yaml --tui). It watches the server's decisions, so the traffic your Python client drives shows up there — the full ops board across eight tabs, plus live binding-axis attribution (which of rate / concurrency / cost bound each denial) for unified policies. Prefer to read it from Python? The MonitorBackend reads the same state over the server's Monitor door — see Fleet & Monitor clients. (Headless or in production? Emit OpenTelemetry → Grafana instead — see the Monitoring guide.)

The direct door — RedisBackend

Configure a strategy and point it at the Redis your fleet shares. The decision is produced server-side, in Lua (the core's own vendored script) — check is the whole surface.

import redis
from throttlekit import RedisBackend, Gcra

client = redis.Redis.from_url("redis://localhost:6379")
api = RedisBackend(client, Gcra(limit=100, period_ms=60_000, burst=20), prefix="prod")

d = api.check(api_key)                  # defaults to the Redis server clock (skew-free across a fleet)

Strategies: Gcra, TokenBucket, FixedWindow, SlidingWindow, SlidingWindowLog. The prefix joins as f"{prefix}:{key}" — the same key scheme the core uses, so a Python and a Node client on one limit address the same Redis key. The backend is client-agnostic: pass any object with evalsha / eval (redis-py satisfies it structurally).

The Decision

Both backends return a frozen Decision dataclass — the five fields of the core's decision, on the wire:

Field	Meaning
allowed	whether the request is permitted
limit	the effective ceiling (burst capacity or window quota)
remaining	whole units left before the next rejection (never negative)
reset_at	epoch-ms at which the limiter is fully replenished
retry_after_ms	ms to wait before retrying (0 when allowed)

ServiceBackend also exposes Forecast (from forecast) and Admission (from admit — see The axes).

Errors

A denial is not an error — it's a Decision with allowed is False. Exceptions are reserved for operational faults, all subclasses of ThrottleKitError:

Exception	When
PolicyNotFoundError	the named policy isn't configured in the service (gRPC NOT_FOUND)
OperationNotSupportedError	wrong op for the policy kind — e.g. admit on a rate limiter, check on a meter (gRPC UNIMPLEMENTED)
ServiceUnavailableError	the service is unreachable (gRPC UNAVAILABLE)

from throttlekit.errors import PolicyNotFoundError

try:
    rl.check("typo-policy", key)
except PolicyNotFoundError:
    ...

Next: The axes — reaching cost, two-tier, and concurrency from Python.