throttled-py

https://github.com/zhuozhuocrayon/throttled-py
Admin
throttled-py is a high-performance Python rate limiting library offering multiple algorithms (Fixed...
Tokens:21,120
Snippets:221
Trust Score:8.5
License:MIT
Update:2 months ago
Show doc for...
Tokens:
Raw
### Install Project Dependencies

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/CONTRIBUTING.md

Installs all project dependencies using the uv package manager.

```bash
uv sync --group all
```

--------------------------------

### Install optional dependencies

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/installation.md

Commands to install additional features like Redis or OpenTelemetry support.

```shell
$ pip install "throttled-py[redis]"
$ pip install "throttled-py[otel]"
$ pip install "throttled-py[redis,otel]"
```

--------------------------------

### Install throttled-py via pip

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/installation.md

Standard installation command for the package.

```default
$ pip install throttled-py
```

--------------------------------

### Install throttled-py for FastAPI

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/contrib/fastapi.md

Command to install the library with FastAPI support.

```bash
pip install 'throttled-py[fastapi]'
```

--------------------------------

### Branch Creation Commands

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/CONTRIBUTING.md

Example commands for updating the main branch and creating a new feature or documentation branch.

```bash
git checkout main
git pull --ff-only upstream main
git checkout -b feat/260405_quota_parser_dsl
git checkout -b docs/260405_contributing_branch_rules
```

--------------------------------

### Configure Redis Cluster Store

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/quickstart/store-backends.md

Examples for using Redis Cluster storage in both synchronous and asynchronous contexts.

```python
from throttled import RateLimiterType, Throttled, store


@Throttled(
    key="/api/products",
    using=RateLimiterType.TOKEN_BUCKET.value,
    quota="1/m",
    # 🌟 use RedisStore as storage
    store=store.RedisStore(
        server="redis+cluster://:yourpassword@host1:6379,host2:6379",
        # 🌟 Pass any extra kwargs for redis-py Cluster client
        options={"REDIS_CLIENT_KWARGS": {}},
    ),
)
def products() -> list[dict[str, str]]:
    return [{"name": "iPhone"}, {"name": "MacBook"}]


def demo() -> None:
    products()  # type: ignore[call-arg]
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    products()  # type: ignore[call-arg]


if __name__ == "__main__":
    demo()
```

```python
import asyncio

from throttled.asyncio import RateLimiterType, Throttled, store


@Throttled(
    key="/api/products",
    using=RateLimiterType.TOKEN_BUCKET.value,
    quota="1/m",
    # 🌟 use RedisStore as storage
    store=store.RedisStore(
        server="redis+cluster://:yourpassword@host1:6379,host2:6379",
        # 🌟 Pass any extra kwargs for redis-py Cluster client
        options={"REDIS_CLIENT_KWARGS": {}},
    ),
)
async def products() -> list[dict[str, str]]:
    return [{"name": "iPhone"}, {"name": "MacBook"}]


async def demo() -> None:
    await products()
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    await products()


if __name__ == "__main__":
    asyncio.run(demo())
```

--------------------------------

### Configure Pre-commit Hooks

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/CONTRIBUTING.md

Commands to install and verify pre-commit hooks managed by prek.

```bash
# Install git hooks (first time only)
uv run prek install
```

```bash
# Should list: check-merge-conflict, ruff, ruff-format, mypy-strict
uv run prek list
```

--------------------------------

### Install throttled-py with OpenTelemetry support

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/observability/opentelemetry.md

Install the package with the otel extra to include the necessary OpenTelemetry API dependencies.

```bash
pip install 'throttled-py[otel]'
```

--------------------------------

### Commit Message Examples

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/CONTRIBUTING.md

Examples of valid commit messages following Conventional Commits and examples of invalid messages that will fail CI.

```bash
# Good
git commit -m "refactor: narrow memory store lock typing"
git commit -m "feat: add hook system with OpenTelemetry support (#37)"
git commit -m "fix: add @wraps decorator for metadata preservation (#120)"
git commit -m "docs: update contributing guidelines (#127)"

# Bad - will fail CI
git commit -m "updated hooks"           # missing type
git commit -m "Feat: add hooks"         # uppercase type
git commit -m "feat : add hooks"        # space before colon
```

--------------------------------

### Registering multiple hooks

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/advance_usage/hooks.md

Example of configuring a Throttled instance with a list of multiple hooks.

```python
throttle = Throttled(
    key="/api",
    quota=per_sec(100),
    hooks=[
        TimingHook(),
        LoggingHook(),
        MetricsHook(statsd_client),
    ],
)
```

--------------------------------

### Async RedisStore Usage

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/quickstart/store-backends.md

Example of using RedisStore in an asynchronous environment with a standalone Redis server.

```python
import asyncio

from throttled.asyncio import RateLimiterType, Throttled, store


@Throttled(
    key="/api/products",
    using=RateLimiterType.TOKEN_BUCKET.value,
    quota="1/m",
    # 🌟 use RedisStore as storage
    store=store.RedisStore(
        server="redis://127.0.0.1:6379/0",
        # 🌟 Pass any extra kwargs for redis-py client.
        options={"REDIS_CLIENT_KWARGS": {}, "CONNECTION_POOL_KWARGS": {}},
    ),
)
async def products() -> list[dict[str, str]]:
    return [{"name": "iPhone"}, {"name": "MacBook"}]


async def demo() -> None:
    await products()
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    await products()


if __name__ == "__main__":
    asyncio.run(demo())
```

--------------------------------

### Sync RedisStore Usage

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/quickstart/store-backends.md

Example of using RedisStore in a synchronous environment with a standalone Redis server.

```python
from throttled import RateLimiterType, Throttled, store


@Throttled(
    key="/api/products",
    using=RateLimiterType.TOKEN_BUCKET.value,
    quota="1/m",
    # 🌟 use RedisStore as storage
    store=store.RedisStore(
        server="redis://127.0.0.1:6379/0",
        # 🌟 Pass any extra kwargs for redis-py client.
        options={"REDIS_CLIENT_KWARGS": {}, "CONNECTION_POOL_KWARGS": {}},
    ),
)
def products() -> list[dict[str, str]]:
    return [{"name": "iPhone"}, {"name": "MacBook"}]


def demo() -> None:
    products()  # type: ignore[call-arg]
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    products()  # type: ignore[call-arg]


if __name__ == "__main__":
    demo()
```

--------------------------------

### Sync Redis Sentinel Usage

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/quickstart/store-backends.md

Example of using RedisStore with a Redis Sentinel configuration.

```python
from throttled import RateLimiterType, Throttled, store


@Throttled(
    key="/api/products",
    using=RateLimiterType.TOKEN_BUCKET.value,
    quota="1/m",
    # 🌟 use RedisStore as storage
    store=store.RedisStore(
        server="redis+sentinel://:yourpassword@host1:26379,host2:26379/mymaster",
        # 🌟 Pass any extra kwargs for redis-py Sentinel client.
        options={
            "SENTINEL_KWARGS": {},
            "REDIS_CLIENT_KWARGS": {},
            "CONNECTION_POOL_KWARGS": {},
        },
    ),
)
def products() -> list[dict[str, str]]:
    return [{"name": "iPhone"}, {"name": "MacBook"}]


def demo() -> None:
    products()  # type: ignore[call-arg]
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    products()  # type: ignore[call-arg]


if __name__ == "__main__":
    demo()
```

--------------------------------

### Hook Type Validation

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/advance_usage/hooks.md

Examples of correct and incorrect hook initialization for sync and async Throttled instances.

```python
from throttled import Throttled, Hook, per_sec
from throttled.asyncio import Throttled as AsyncThrottled
from throttled.asyncio.hooks import Hook as AsyncHook

# ✅ Correct: sync Hook with sync Throttled
Throttled(key="/api", quota=per_sec(10), hooks=[MySyncHook()])

# ✅ Correct: async Hook with async Throttled
AsyncThrottled(key="/api", quota=per_sec(10), hooks=[MyAsyncHook()])

# ❌ TypeError: async Hook with sync Throttled
Throttled(key="/api", quota=per_sec(10), hooks=[MyAsyncHook()])

# ❌ TypeError: non-hook object
Throttled(key="/api", quota=per_sec(10), hooks=["not a hook"])
```

--------------------------------

### Build and Preview Documentation

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/CONTRIBUTING.md

Commands to build the Sphinx documentation locally and serve the generated HTML files for preview.

```bash
# Build HTML docs (sphinx is included in `uv sync --group all`)
cd docs
uv run make html

# Preview locally at http://localhost:8000
python -m http.server 8000 --directory build/html
```

--------------------------------

### Initialize MemoryStore

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/advance_usage/store-configuration.md

Configure the MemoryStore with a custom maximum size for both sync and async environments.

```python
from throttled import store

store.MemoryStore(options={"MAX_SIZE": 10240})
```

```python
from throttled.asyncio import store

store.MemoryStore(options={"MAX_SIZE": 10240})
```

--------------------------------

### Using Global MemoryStore for Sync and Async

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/quickstart/store-backends.md

Demonstrates using the default global MemoryStore for rate limiting synchronous and asynchronous functions.

```python
from throttled import Throttled


# 🌟 Use the global MemoryStore as the storage backend.
@Throttled(key="/api/products", quota="1/m")
def products() -> list[dict[str, str]]:
    return [{"name": "iPhone"}, {"name": "MacBook"}]


def demo() -> None:
    products()  # type: ignore[call-arg]
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    products()  # type: ignore[call-arg]


if __name__ == "__main__":
    demo()
```

```python
import asyncio

from throttled.asyncio import Throttled


# 🌟 Use the global MemoryStore as the storage backend.
@Throttled(key="/api/products", quota="1/m")
async def products() -> list[dict[str, str]]:
    return [{"name": "iPhone"}, {"name": "MacBook"}]


async def demo() -> None:
    await products()
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    await products()


if __name__ == "__main__":
    asyncio.run(demo())
```

--------------------------------

### Configure Throttled with Quota Strings

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/quickstart/quota-configuration.md

Demonstrates initializing synchronous and asynchronous Throttled instances using readable quota strings.

```python
from throttled import Throttled

throttle = Throttled(
    key="/api/ping",
    quota="100/s",
    # quota="100/s burst 200",
    # quota="100 per second",
    # quota="100 per second burst 200",
)


if __name__ == "__main__":
    print(throttle.limit())
```

```python
import asyncio

from throttled.asyncio import Throttled

throttle = Throttled(
    key="/api/ping",
    quota="100/s",
    # quota="100/s burst 200",
    # quota="100 per second",
    # quota="100 per second burst 200",
)


async def main() -> None:
    print(await throttle.limit())


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Initialize RedisStore

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/advance_usage/store-configuration.md

Configure RedisStore for synchronous or asynchronous operations using a Redis URL.

```python
from throttled import store

store.RedisStore(server="redis://127.0.0.1:6379/0", options={})
```

```python
from throttled.asyncio import store

store.RedisStore(server="redis://127.0.0.1:6379/0", options={})
```

--------------------------------

### MemoryStore Initialization

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/advance_usage/store-configuration.md

Initializes a MemoryStore instance for either synchronous or asynchronous usage with custom options.

```APIDOC
## MemoryStore Initialization

### Description
Initializes the MemoryStore, which acts as an LRU cache with expiration. It is available for both synchronous and asynchronous contexts.

### Constructor
`throttled.store.MemoryStore(options)` or `throttled.asyncio.store.MemoryStore(options)`

### Parameters
- **options** (dict) - Optional - Configuration dictionary for the store.

### Options
- **MAX_SIZE** (int) - Optional - Maximum capacity of the cache. Defaults to 1024. When exceeded, items are removed based on the LRU policy.

### Example
```python
from throttled import store

# Initialize with custom size
my_store = store.MemoryStore(options={"MAX_SIZE": 10240})
```
```

--------------------------------

### Run test suites

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/CONTRIBUTING.md

Commands for executing the full test suite or specific test files using uv.

```bash
# Run all tests
uv run pytest -n auto tests/ -x

# Run a specific test file
uv run pytest -n auto tests/test_hooks.py -v
```

--------------------------------

### Implement per-route quotas

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/contrib/fastapi.md

Demonstrates how to apply different rate limits to different routes using a shared memory store.

```python
from fastapi import FastAPI, Request
from throttled.asyncio.contrib.fastapi import (
    Limiter,
    RateLimitExceededError,
    RateLimitMiddleware,
    rate_limit_exceeded_handler,
)
from throttled.asyncio.store import MemoryStore

# 1) Create a limiter with a shared store for all routes.
store = MemoryStore()
limiter = Limiter("10/m", store=store)

# 2) Wire FastAPI integration hooks:
#    middleware adds RateLimit-* headers
#    handler renders HTTP 429 responses.
app = FastAPI()
app.add_middleware(RateLimitMiddleware)
app.add_exception_handler(RateLimitExceededError, rate_limit_exceeded_handler)


# 3) Apply the default quota to one route.
@app.get("/items")
@limiter.limit()
async def list_items(request: Request) -> dict[str, list[str]]:
    return {"items": ["apple", "banana"]}


# 4) Override the quota for another route.
@app.get("/admin")
@limiter.limit("1/m")
async def admin_panel(request: Request) -> dict[str, str]:
    return {"status": "ok"}
```

--------------------------------

### Configure Redis Sentinel Store (Async)

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/quickstart/store-backends.md

Uses RedisStore with a Sentinel connection string and custom client options for asynchronous functions.

```python
import asyncio

from throttled.asyncio import RateLimiterType, Throttled, store


@Throttled(
    key="/api/products",
    using=RateLimiterType.TOKEN_BUCKET.value,
    quota="1/m",
    # 🌟 use RedisStore as storage
    store=store.RedisStore(
        server="redis+sentinel://:yourpassword@host1:26379,host2:26379/mymaster",
        # 🌟 Pass any extra kwargs for redis-py Sentinel client.
        options={
            "SENTINEL_KWARGS": {},
            "REDIS_CLIENT_KWARGS": {},
            "CONNECTION_POOL_KWARGS": {},
        },
    ),
)
async def products() -> list[dict[str, str]]:
    return [{"name": "iPhone"}, {"name": "MacBook"}]


async def demo() -> None:
    await products()
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    await products()


if __name__ == "__main__":
    asyncio.run(demo())
```

--------------------------------

### RedisStore Initialization

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/advance_usage/store-configuration.md

Initializes a RedisStore instance for managing rate limiting state in Redis.

```APIDOC
## RedisStore

### Description
Initializes the RedisStore backend for throttled. Supports both synchronous and asynchronous usage.

### Constructor
`throttled.store.RedisStore(server: str, options: dict)`
`throttled.asyncio.store.RedisStore(server: str, options: dict)`

### Parameters
- **server** (str) - Optional - Standard Redis URL. Default: "redis://localhost:6379/0"
- **options** (dict) - Optional - Configuration options for the Redis connection. Default: {}

### Options
- **SOCKET_TIMEOUT** (float) - Optional - ConnectionPool timeout.
- **SOCKET_CONNECT_TIMEOUT** (float) - Optional - ConnectionPool connect timeout.
- **CONNECTION_POOL_KWARGS** (dict) - Optional - Additional parameters for ConnectionPool construction.
```

--------------------------------

### Configure Redis Storage Backend

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/README.md

Use RedisStore to handle rate limiting with standalone, sentinel, or cluster modes. The options parameter accepts standard Redis configuration items.

```python
from throttled import RateLimiterType, Throttled, store

@Throttled(
    key="/api/products",
    using=RateLimiterType.TOKEN_BUCKET.value,
    quota="1/m",
    store=store.RedisStore(
        # Standalone mode
        server="redis://127.0.0.1:6379/0",
        # Sentinel mode
        # server="redis+sentinel://:yourpassword@host1:26379,host2:26379/mymaster"
        # Cluster mode
        # server="redis+cluster://:yourpassword@host1:6379,host2:6379",
        options={}
    ),
)
def products() -> list:
    return [{"name": "iPhone"}, {"name": "MacBook"}]

products()  # Success
products()  # Raises LimitedError
```

--------------------------------

### Configure In-Memory Storage Backend

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/README.md

Use a shared MemoryStore instance to maintain consistent rate limiting across different functions. If no store is specified, a default global instance is used.

```python
from throttled import Throttled, store

mem_store = store.MemoryStore()

@Throttled(key="ping-pong", quota="1/m", store=mem_store)
def ping() -> str: return "ping"

@Throttled(key="ping-pong", quota="1/m", store=mem_store)
def pong() -> str: return "pong"

ping()  # Success
pong()  # Raises LimitedError
```

--------------------------------

### Perform Basic Rate Limiting

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/README.md

Demonstrates manual rate limiting using the Throttled class with default in-memory storage.

```python
from throttled import Throttled

# Default: In-Memory storage, Token Bucket algorithm, 60 reqs / min.
throttle = Throttled()

# Deduct 1 request -> RateLimitResult(limited=False,
# state=RateLimitState(limit=60, remaining=59, reset_after=1, retry_after=0))
print(throttle.limit("key", 1))
# Check state -> RateLimitState(limit=60, remaining=59, reset_after=1, retry_after=0)
print(throttle.peek("key"))

# Deduct 60 requests (limited) -> RateLimitResult(limited=False,
# state=RateLimitState(limit=60, remaining=59, reset_after=1, retry_after=60))
print(throttle.limit("key", 60))
```

--------------------------------

### Test per-route rate limits with curl

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/contrib/fastapi.md

Demonstrates verifying rate limit behavior for different routes using curl commands.

```bash
$ curl -is http://localhost:8000/items
HTTP/1.1 200 OK
ratelimit-limit: 10
ratelimit-remaining: 9
...

$ curl -is http://localhost:8000/admin
HTTP/1.1 200 OK
ratelimit-limit: 1
ratelimit-remaining: 0
...
{"status":"ok"}

$ curl -is http://localhost:8000/admin
HTTP/1.1 429 Too Many Requests
ratelimit-limit: 1
retry-after: 60
...
{"detail":"Rate limit exceeded"}

$ curl -is http://localhost:8000/items
HTTP/1.1 200 OK
ratelimit-limit: 10
ratelimit-remaining: 8
...
```

--------------------------------

### Sharing a Custom MemoryStore Instance

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/quickstart/store-backends.md

Shows how to share a specific MemoryStore instance between multiple functions to enforce a unified rate limit.

```python
from throttled import Throttled, store

# 🌟 Use MemoryStore as the storage backend.
mem_store = store.MemoryStore()


@Throttled(key="ping-pong", quota="1/m", store=mem_store)
def ping() -> str:
    return "ping"


@Throttled(key="ping-pong", quota="1/m", store=mem_store)
def pong() -> str:
    return "pong"


def demo() -> None:
    # >> ping
    ping()  # type: ignore[call-arg]
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    pong()  # type: ignore[call-arg]


if __name__ == "__main__":
    demo()
```

```python
import asyncio

from throttled.asyncio import Throttled, store

# 🌟 Use MemoryStore as the storage backend.
mem_store = store.MemoryStore()


@Throttled(key="ping-pong", quota="1/m", store=mem_store)
async def ping() -> str:
    return "ping"


@Throttled(key="ping-pong", quota="1/m", store=mem_store)
async def pong() -> str:
    return "pong"


async def demo() -> None:
    # >> ping
    await ping()
    # >> throttled.exceptions.LimitedError:
    # Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
    await pong()


if __name__ == "__main__":
    asyncio.run(demo())
```

--------------------------------

### Configure OTelHook for Synchronous and Asynchronous Applications

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/observability/opentelemetry.md

Initialize OTelHook by providing a Meter instance from the OpenTelemetry API. The same pattern applies to both sync and async implementations.

```python
from opentelemetry import metrics
from throttled.contrib.otel import OTelHook

meter = metrics.get_meter("my-service", "1.0.0")
hook = OTelHook(meter)
```

```python
from opentelemetry import metrics
from throttled.asyncio.contrib.otel import OTelHook

meter = metrics.get_meter("my-service", "1.0.0")
hook = OTelHook(meter)
```

--------------------------------

### Configure OTelHook for synchronous rate limiting

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/observability/opentelemetry.md

Initialize OTelHook with an OpenTelemetry meter and attach it to a synchronous Throttled instance.

```python
from opentelemetry.metrics import get_meter
from throttled import Throttled
from throttled.contrib.otel import OTelHook

# Create OTelHook with a meter from the OTel API.
# The actual metrics backend (e.g., Prometheus, OTLP) is configured separately
# via opentelemetry-sdk or your framework's setup.
meter = get_meter("throttled-example")

# Create a rate limiter with OTelHook attached.
throttle = Throttled(key="/api/ping", quota="5/m", hooks=[OTelHook(meter)])


def main() -> None:
    # First 5 requests are allowed.
    for i in range(5):
        result = throttle.limit("/api/ping")
        print(f"Request {i + 1}: {'denied' if result.limited else 'allowed'}")

    # But 6th request is denied.
    result = throttle.limit("/api/ping")
    print(f"Request 6: {'denied' if result.limited else 'allowed'}")

    # 📊 OTelHook records the following metrics:
    #   - throttled.requests (Counter): number of rate limit checks
    #   - throttled.duration (Histogram): latency of rate limit checks
    # Attributes: key, algorithm, store_type, result ("allowed" / "denied")


if __name__ == "__main__":
    main()
```

--------------------------------

### Run Pre-commit Hooks Manually

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/CONTRIBUTING.md

Executes lint and format checks manually on specific files or the entire repository.

```bash
# Run on specific files
uv run prek run --files <changed-files>

# Run on all files
uv run prek run --all-files
```

--------------------------------

### Configure OTelHook for asynchronous rate limiting

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/observability/opentelemetry.md

Initialize OTelHook with an OpenTelemetry meter and attach it to an asynchronous Throttled instance.

```python
import asyncio

from opentelemetry.metrics import get_meter
from throttled.asyncio import Throttled
from throttled.asyncio.contrib.otel import OTelHook

# Create OTelHook with a meter from the OTel API.
# The actual metrics backend (e.g., Prometheus, OTLP) is configured separately
# via opentelemetry-sdk or your framework's setup.
meter = get_meter("throttled-example")

# Create an async rate limiter with OTelHook attached.
throttle = Throttled(key="/api/ping", quota="5/m", hooks=[OTelHook(meter)])


async def main() -> None:
    # First 5 requests are allowed.
    for i in range(5):
        result = await throttle.limit("/api/ping")
        print(f"Request {i + 1}: {'denied' if result.limited else 'allowed'}")

    # But 6th request is denied.
    result = await throttle.limit("/api/ping")
    print(f"Request 6: {'denied' if result.limited else 'allowed'}")

    # 📊 OTelHook records the following metrics:
    #   - throttled.requests (Counter): number of rate limit checks
    #   - throttled.duration (Histogram): latency of rate limit checks
    # Attributes: key, algorithm, store_type, result ("allowed" / "denied")


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Configure OTLP Metric Exporter

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/observability/opentelemetry.md

Use this configuration to send metrics to an OTLP-compatible collector via gRPC.

```python
from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter

reader = PeriodicExportingMetricReader(
    OTLPMetricExporter(endpoint="http://collector:4317")
)
provider = MeterProvider(metric_readers=[reader])
metrics.set_meter_provider(provider)
```

--------------------------------

### Branch Naming Format

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/CONTRIBUTING.md

The required format for creating new branches from main.

```text
<type>/<yymmdd>_<topic>
```

--------------------------------

### Implement a synchronous logging hook

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/advance_usage/hooks.md

Demonstrates creating a custom Hook class for synchronous rate limiting to log check status and remaining quota.

```python
from collections.abc import Callable

from throttled import Hook, HookContext, RateLimitResult, Throttled


class LoggingHook(Hook):
    def on_limit(  # noqa: PLR6301
        self,
        call_next: Callable[[], RateLimitResult],
        context: HookContext,
    ) -> RateLimitResult:
        # Before rate limit check
        print(f"Checking rate limit for {context.key}")

        # Execute rate limit check
        result = call_next()

        # After rate limit check
        status = "denied" if result.limited else "allowed"
        print(f"[{context.key}] {status} - remaining: {result.state.remaining}")

        return result


throttle = Throttled(key="/api/users", quota="10/s", hooks=[LoggingHook()])


def main() -> None:
    result = throttle.limit()
    # Output:
    # Checking rate limit for /api/users
    # [/api/users] allowed - remaining: 9
    print(f"limited={result.limited}")


if __name__ == "__main__":
    main()
```

--------------------------------

### Thebe configuration

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/installation.md

Configuration block for the Thebe interactive environment.

```json
{
    requestKernel: true,
    binderOptions: {
        repo: "binder-examples/jupyter-stacks-datascience",
        ref: "master",
    },
    codeMirrorConfig: {
        theme: "abcdef",
        mode: "python"
    },
    kernelOptions: {
        name: "python3",
        path: "./."
    },
    predefinedOutput: true
}
```

--------------------------------

### store.RedisStore

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/README.md

Configures a Redis backend for storing rate limit state, supporting standalone, sentinel, and cluster modes.

```APIDOC
## store.RedisStore

### Description
Initializes a Redis storage backend for the Throttled decorator.

### Parameters
- **server** (str) - Required - Connection string (e.g., "redis://127.0.0.1:6379/0").
- **options** (dict) - Optional - Additional Redis configuration items.
```

--------------------------------

### Specify Rate Limiting Algorithm (Sync)

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/quickstart/specifying-algorithms.md

Configures a synchronous Throttled instance with a specific algorithm and quota.

```python
from throttled import RateLimiterType, Throttled


def main() -> None:
    throttle = Throttled(
        # 🌟Specifying a current limiting algorithm
        using=RateLimiterType.FIXED_WINDOW.value,
        # using=RateLimiterType.SLIDING_WINDOW.value,
        # using=RateLimiterType.LEAKING_BUCKET.value,
        # using=RateLimiterType.TOKEN_BUCKET.value,
        # using=RateLimiterType.GCRA.value,
        quota="1/m",
    )
    assert throttle.limit("key", 2).limited


if __name__ == "__main__":
    main()
```

--------------------------------

### Specify Rate Limiting Algorithm (Async)

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/quickstart/specifying-algorithms.md

Configures an asynchronous Throttled instance with a specific algorithm and quota.

```python
import asyncio

from throttled.asyncio import RateLimiterType, Throttled


async def main() -> None:
    throttle = Throttled(
        # 🌟Specifying a current limiting algorithm
        using=RateLimiterType.FIXED_WINDOW.value,
        # using=RateLimiterType.SLIDING_WINDOW.value,
        # using=RateLimiterType.LEAKING_BUCKET.value,
        # using=RateLimiterType.TOKEN_BUCKET.value,
        # using=RateLimiterType.GCRA.value,
        quota="1/m",
    )
    assert (await throttle.limit("key", 2)).limited


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Basic hook structure

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/advance_usage/hooks.md

Demonstrates the mandatory requirement to call call_next() within the on_limit method.

```python
class MyHook(Hook):
    def on_limit(self, call_next, context):
        # Must call call_next() and return its result
        result = call_next()
        return result
```

--------------------------------

### Implement an asynchronous logging hook

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/advance_usage/hooks.md

Demonstrates creating an asynchronous Hook class for non-blocking rate limiting operations.

```python
import asyncio
from collections.abc import Awaitable, Callable

from throttled.asyncio import Hook, HookContext, RateLimitResult, Throttled


class LoggingHook(Hook):
    async def on_limit(  # noqa: PLR6301
        self,
        call_next: Callable[[], Awaitable[RateLimitResult]],
        context: HookContext,
    ) -> RateLimitResult:
        # Before rate limit check
        print(f"Checking rate limit for {context.key}")

        # Execute rate limit check
        result = await call_next()

        # After rate limit check
        status = "denied" if result.limited else "allowed"
        print(f"[{context.key}] {status} - remaining: {result.state.remaining}")

        return result


throttle = Throttled(key="/api/users", quota="10/s", hooks=[LoggingHook()])


async def main() -> None:
    result = await throttle.limit()
    # Output:
    # Checking rate limit for /api/users
    # [/api/users] allowed - remaining: 9
    print(f"limited={result.limited}")


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Define Custom Quota Objects

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/quickstart/quota-configuration.md

Shows how to programmatically build Quota objects using built-in constructors or custom durations when string patterns are insufficient.

```python
from datetime import timedelta

from throttled import rate_limiter

# Built-in constructors for common fixed periods.
rate_limiter.per_sec(60)  # 60 req/sec
rate_limiter.per_min(60)  # 60 req/min
rate_limiter.per_hour(60)  # 60 req/hour
rate_limiter.per_day(60)  # 60 req/day
rate_limiter.per_week(60)  # 60 req/week

# Allow up to 120 requests as burst capacity.
# If burst is omitted, it defaults to the same value as limit.
rate_limiter.per_min(60, burst=120)

# Custom period example:
# allow 120 requests per 2 minutes, with burst capacity of 150.
rate_limiter.per_duration(timedelta(minutes=2), limit=120, burst=150)
```

--------------------------------

### PromQL Queries for Grafana Dashboards

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/observability/opentelemetry.md

Use these queries to visualize request metrics, denial rates, and latency distributions in Grafana.

```promql
# Request rate by key
sum(rate(throttled_requests_total[5m])) by (key)

# Denial rate
sum(rate(throttled_requests_total{result="denied"}[5m]))
/
sum(rate(throttled_requests_total[5m]))

# Allowed vs Denied comparison
sum(rate(throttled_requests_total{result="allowed"}[5m]))
sum(rate(throttled_requests_total{result="denied"}[5m]))

# P95 duration by algorithm
histogram_quantile(0.95, rate(throttled_duration_bucket[5m])) by (algorithm)
```

--------------------------------

### Define Custom Key Functions

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/contrib/fastapi.md

Implement sync or async functions to extract rate-limit keys from request headers or authentication tokens.

```python
# Sync: simple header extraction.
def get_api_key(request: Request) -> str:
    return request.headers.get("X-API-Key", "anonymous")

# Async: database or Redis lookup.
async def get_user_id(request: Request) -> str:
    token = request.headers.get("Authorization", "")
    user = await verify_token(token)
    return user.id
```

--------------------------------

### Configure Quota Settings

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/README.md

Define rate limits using human-readable strings supporting units like seconds, minutes, hours, days, and weeks, with optional burst capacity.

```python
from throttled import Throttled

throttle = Throttled(
    key="/api/ping",
    quota="100/s",
    # quota="100/s burst 200",
    # quota="100 per second",
    # quota="100 per second burst 200",
)


if __name__ == "__main__":
    print(throttle.limit())
```

--------------------------------

### Run local checks

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/CONTRIBUTING.md

Commands to execute linting and test suites locally before submitting a pull request.

```bash
uv run prek run --all-files
uv run pytest -n auto tests/ -x
```

--------------------------------

### Safe hook implementation with error handling

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/advance_usage/hooks.md

Shows how to wrap hook logic in try/except blocks to prevent the hook from being skipped if an error occurs.

```python
class SafeHook(Hook):
    def on_limit(self, call_next, context):
        # Pre-processing with error handling
        try:
            self.validate_request(context)
        except Exception as e:
            logging.warning(f"Pre-processing failed: {e}")

        # call_next() outside try/except - always executes
        result = call_next()

        # Post-processing with error handling
        try:
            self.send_metrics(result, context)
        except Exception as e:
            logging.warning(f"Post-processing failed: {e}")

        return result
```

--------------------------------

### Implement Synchronous Rate Limiting

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/quickstart/index.md

Uses the Token Bucket algorithm with the default MemoryStore to limit requests synchronously.

```python
from throttled import RateLimiterType, Throttled, utils

throttle = Throttled(
    # 📈 Use Token Bucket algorithm.
    using=RateLimiterType.TOKEN_BUCKET.value,
    # 🪣 Set quota: 1,000 tokens per second (limit), bucket size 1,000 (burst).
    quota="1000/s burst 1000",
    # 📁 By default, global MemoryStore is used as the storage backend.
)


def call_api() -> bool:
    # 💧 Deduct 1 token for key="/ping"
    result = throttle.limit("/ping", cost=1)
    return result.limited


if __name__ == "__main__":
    # 💻 Python 3.12.10, Linux 5.4.119-1-tlinux4-0009.1, Arch: x86_64, Specs: 2C4G.
    # ✅ Total: 100000, 🕒 Latency: 0.0068 ms/op, 🚀 Throughput: 122513 req/s (--)
    # ❌ Denied: 98000 requests
    benchmark: utils.Benchmark = utils.Benchmark()
    denied_num: int = sum(benchmark.serial(call_api, 100_000))
    print(f"❌ Denied: {denied_num} requests")
```

--------------------------------

### Asynchronous Rate Limiting

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/quickstart/function-call.md

Demonstrates basic usage of the asynchronous Throttled class, requiring await for limit and peek operations.

```python
import asyncio

from throttled.asyncio import Throttled


async def main() -> None:
    # By Default, it initializes a rate limiter with In-Memory,
    # allowing 60 requests per minute, using the token bucket algorithm.
    # Default: In-Memory storage, Token Bucket algorithm, 60 reqs / min.
    throttle = Throttled()

    # Consume 1 token.
    result = await throttle.limit("key")
    # Should not be limited.
    assert not result.limited

    # Get the state of the rate limiter:
    # >> RateLimitState(limit=60, remaining=59, reset_after=1, retry_after=0))
    print(result.state)

    # You can also get the state by using the `peek` method.
    # >> RateLimitState(limit=60, remaining=59, reset_after=1, retry_after=0)
    print(await throttle.peek("key"))

    # You can also specify the cost of the request.
    result = await throttle.limit("key", cost=60)
    # This will consume 60 tokens, which exceeds the limit of 60 tokens per minute.
    assert result.limited

    # >> RateLimitState(limit=60, remaining=59, reset_after=1, retry_after=1))
    print(result.state)


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Configure Console Metric Exporter

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/observability/opentelemetry.md

Use this configuration for debugging purposes to output metrics directly to the console.

```python
from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import ConsoleMetricExporter, PeriodicExportingMetricReader

reader = PeriodicExportingMetricReader(ConsoleMetricExporter())
provider = MeterProvider(metric_readers=[reader])
metrics.set_meter_provider(provider)

# Now get_meter() will use this provider
meter = metrics.get_meter("my-app")
```

--------------------------------

### Apply test naming conventions

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/CONTRIBUTING.md

Use the format test_{function_name}__{case_description} with double underscores to separate the target function from the specific test case.

```python
class TestOTelHook:
    def test_allowed_request__records_metrics(cls, ...): ...
    def test_denied_request__records_metrics(cls, ...): ...
    def test_custom_cost__recorded(cls, ...): ...
```

--------------------------------

### Configure Prometheus Metric Exporter

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/observability/opentelemetry.md

Use this configuration to expose metrics via an HTTP server for Prometheus scraping.

```python
from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.exporter.prometheus import PrometheusMetricReader
from prometheus_client import start_http_server

start_http_server(port=8000)
reader = PrometheusMetricReader()
provider = MeterProvider(metric_readers=[reader])
metrics.set_meter_provider(provider)

# Metrics available at http://localhost:8000/metrics
```

--------------------------------

### store.MemoryStore

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/README.md

An in-memory storage backend for rate limiting. Useful for single-instance applications.

```APIDOC
## store.MemoryStore

### Description
Initializes an in-memory storage backend. If multiple Throttled instances share the same MemoryStore instance, they will share the same rate limit state.
```

--------------------------------

### Synchronous Rate Limiting

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/quickstart/function-call.md

Demonstrates basic usage of the synchronous Throttled class, including limiting requests, checking state, and using the peek method.

```python
from throttled import Throttled


def main() -> None:
    # By Default, it initializes a rate limiter with In-Memory,
    # allowing 60 requests per minute, using the token bucket algorithm.
    # Default: In-Memory storage, Token Bucket algorithm, 60 reqs / min.
    throttle = Throttled()

    # Consume 1 token.
    result = throttle.limit("key")
    # Should not be limited.
    assert not result.limited

    # Get the state of the rate limiter:
    # >> RateLimitState(limit=60, remaining=59, reset_after=1, retry_after=0))
    print(result.state)

    # You can also get the state by using the `peek` method.
    # >> RateLimitState(limit=60, remaining=59, reset_after=1, retry_after=0)
    print(throttle.peek("key"))

    # You can also specify the cost of the request.
    result = throttle.limit("key", cost=60)
    # This will consume 60 tokens, which exceeds the limit of 60 tokens per minute.
    assert result.limited

    # >> RateLimitState(limit=60, remaining=59, reset_after=1, retry_after=1))
    print(result.state)


if __name__ == "__main__":
    main()
```

--------------------------------

### Redis URI Formats

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/quickstart/store-backends.md

Supported URI schemes for connecting to standalone Redis instances.

```default
redis://[[username]:[password]@]localhost:6379/0
rediss://[[username]:[password]@]localhost:6379/0
unix://[username@]/path/to/socket.sock?db=0[&password=password]
```

--------------------------------

### Test API Key Quota Isolation

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/contrib/fastapi.md

Verify that different API keys maintain independent rate limit quotas.

```bash
$ curl -is -H "X-API-Key: user-a" http://localhost:8000/items
HTTP/1.1 200 OK
ratelimit-remaining: 1
...

$ curl -is -H "X-API-Key: user-a" http://localhost:8000/items
HTTP/1.1 200 OK
ratelimit-remaining: 0
...

$ curl -is -H "X-API-Key: user-a" http://localhost:8000/items
HTTP/1.1 429 Too Many Requests
retry-after: 30
...
{"detail":"Rate limit exceeded"}

$ curl -is -H "X-API-Key: user-b" http://localhost:8000/items
HTTP/1.1 200 OK
ratelimit-remaining: 1
...
{"items":["apple","banana"]}
```

--------------------------------

### Rebase branch onto main

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/CONTRIBUTING.md

Commands to update your local branch with the latest changes from the main branch.

```bash
# Option A: if you have an upstream remote
git fetch upstream main
git rebase upstream/main

# Option B: if you only have origin
git fetch origin main
git rebase origin/main
```

--------------------------------

### Mock external dependencies with pytest fixtures

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/CONTRIBUTING.md

Use MagicMock to simulate external dependencies like the OpenTelemetry SDK to maintain test simplicity.

```python
@pytest.fixture
def mock_meter() -> MagicMock:
    meter: MagicMock = MagicMock(name="Meter")
    meter.create_counter.return_value = MagicMock(name="Counter")
    meter.create_histogram.return_value = MagicMock(name="Histogram")
    return meter
```

--------------------------------

### Sync Wait-Retry Function Call

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/quickstart/wait-retry.md

Demonstrates using the timeout parameter in a synchronous function call to wait for available tokens.

```python
import time

from throttled import Throttled, utils


def main() -> None:
    # Allow 1 burst request, producing 1 token per second.
    throttle = Throttled(key="key", quota="1/s burst 1")

    # Consume burst request quota.
    assert not throttle.limit().limited

    timer = utils.Timer(
        clock=time.time,
        callback=lambda elapsed, start, end: print(f"elapsed: {elapsed:.2f} seconds"),
    )
    with timer:
        # Enabled wait-retry, which will wait for the next available token
        # if the limit is reached.
        # > elapsed: 1.00 seconds
        assert not throttle.limit(timeout=1).limited

    with timer:
        # If the timeout is exceeded, it will return the last RateLimitResult.
        # timeout < ``RateLimitResult.retry_after``, return immediately.
        # > elapsed: 0 seconds
        assert throttle.limit(timeout=0.5).limited


if __name__ == "__main__":
    main()
```

--------------------------------

### Push branch changes

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/CONTRIBUTING.md

Standard command to push your branch to the remote repository.

```bash
git push origin <your-branch>
```

--------------------------------

### Asynchronous Rate Limiting with Context Manager

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/quickstart/context-manager.md

Uses the async Throttled context manager to wrap asynchronous code blocks. Requires an async context manager syntax.

```python
import asyncio

from throttled.asyncio import Throttled, exceptions


async def call_api() -> None:
    print("doing something...")


async def main() -> None:
    throttle: Throttled = Throttled(key="/api/v1/users/", quota="1/m")
    async with throttle as result:
        # The first call will not be rate limited.
        assert not result.limited
        # Get the state of the rate limiter:
        # >> RateLimitState(limit=1, remaining=0, reset_after=60, retry_after=0)
        print(result.state)

        await call_api()

    try:
        async with throttle:
            await call_api()
    except exceptions.LimitedError as exc:
        # >> Rate limit exceeded: remaining=0, reset_after=60, retry_after=60.
        print(exc)


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Middleware Execution Order

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/advance_usage/hooks.md

Visual representation of the Chain of Responsibility pattern used by hooks.

```text
hooks = [A, B]

Execution order:
A.on_limit (before) → B.on_limit (before) → rate_limit → B.on_limit (after) → A.on_limit (after)
```

--------------------------------

### Redis Cluster Connection String Format

Source: https://github.com/zhuozhuocrayon/throttled-py/blob/main/docs/source/quickstart/store-backends.md

The required URI format for connecting to a Redis Cluster.

```default
redis+cluster://[[username]:[password]@]host1[:port1][,hostN][:portN]
```
Tokens:
Raw