> ## Documentation Index
> Fetch the complete documentation index at: https://choked.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Choked decorator

# Choked Class

The main class for creating configurable rate limiting decorators with dual limiting support.

## Class Signature

```python theme={null}
class Choked:
    def __init__(self, redis_url: Optional[str] = None, api_token: Optional[str] = None)
    def __call__(self, key: str, request_limit: Optional[str] = None, 
                 token_limit: Optional[str] = None, token_estimator: Optional[str] = None) -> Callable
```

## Constructor Parameters

<ParamField path="redis_url" type="str" optional>
  Redis connection URL for distributed rate limiting. Mutually exclusive with api\_token.

  **Examples:**

  * `"redis://localhost:6379/0"`
  * `"redis://user:pass@host:6379/0"`
  * `"redis://localhost:6379"`
</ParamField>

<ParamField path="api_token" type="str" optional>
  API token for managed rate limiting service. Mutually exclusive with redis\_url.

  Contact us for access to the managed service.
</ParamField>

## Decorator Parameters

When using a Choked instance as a decorator, it accepts these parameters:

<ParamField path="key" type="str" required>
  Unique identifier for the rate limit bucket. Functions with the same key share the same rate limits.

  **Examples:**

  * `"openai_chat"`
  * `"user_123_api"`
  * `"embedding_service"`
</ParamField>

<ParamField path="request_limit" type="str" optional>
  Request rate limit in format 'number/period'. Optional if token\_limit is provided.

  **Format:** `"number/period"` where period is `"s"` (seconds) or `"m"` (minutes)

  **Examples:**

  * `"10/s"` - 10 requests per second
  * `"100/m"` - 100 requests per minute
  * `"1000/s"` - 1000 requests per second
</ParamField>

<ParamField path="token_limit" type="str" optional>
  Token rate limit in format 'number/period'. Optional if request\_limit is provided.

  **Format:** `"number/period"` where period is `"s"` (seconds) or `"m"` (minutes)

  **Examples:**

  * `"1000/s"` - 1000 tokens per second
  * `"100000/m"` - 100,000 tokens per minute
  * `"50000/m"` - 50,000 tokens per minute
</ParamField>

<ParamField path="token_estimator" type="str" optional>
  Token estimation method. Required when using token\_limit.

  **Options:**

  * `"openai"` - Use OpenAI/tiktoken for text estimation
  * `"voyageai"` - Use VoyageAI tokenizer for text estimation
  * `"default"` - Use tiktoken with GPT-4 tokenizer (same as "openai")
</ParamField>

## Returns

The `__call__` method returns a decorator function that can be applied to both synchronous and asynchronous functions.

## Usage Examples

### Basic Request Limiting

```python theme={null}
from choked import Choked

choke = Choked(redis_url="redis://localhost:6379/0")

@choke(key="api_calls", request_limit="10/s")
def make_api_call():
    # This function is rate limited to 10 requests per second
    return "API response"
```

### Token-Only Limiting for AI APIs

```python theme={null}
from choked import Choked

choke = Choked(api_token="your-api-token")

@choke(key="openai_embed", token_limit="1000000/m", token_estimator="openai")
def get_embeddings(texts):
    # Rate limited by estimated tokens only
    return openai.embeddings.create(input=texts, model="text-embedding-3-small")
```

### Dual Limiting

```python theme={null}
from choked import Choked

choke = Choked(redis_url="redis://localhost:6379/0")

@choke(key="gpt4_chat", request_limit="50/s", token_limit="100000/m", token_estimator="openai")
def chat_completion(messages):
    # Limited by both requests (50/s) AND tokens (100K/m)
    return openai.chat.completions.create(
        model="gpt-4",
        messages=messages
    )
```

### Async Function Support

```python theme={null}
import asyncio
from choked import Choked

choke = Choked(redis_url="redis://localhost:6379/0")

@choke(key="async_api", request_limit="5/s")
async def async_api_call(data):
    # Async functions are automatically detected
    await asyncio.sleep(0.1)
    return f"Processed {data}"

async def main():
    result = await async_api_call("test data")
    print(result)

asyncio.run(main())
```

### Multiple Services

```python theme={null}
from choked import Choked

# Different backends for different services
openai_choke = Choked(api_token="openai-service-token")
redis_choke = Choked(redis_url="redis://localhost:6379/0")

@openai_choke(key="gpt4", request_limit="50/s", token_limit="100000/m", token_estimator="openai")
def openai_call(messages):
    return openai.chat.completions.create(model="gpt-4", messages=messages)

@redis_choke(key="internal_api", request_limit="1000/s")
def internal_call():
    return "Internal API response"
```

### Shared Rate Limits

```python theme={null}
from choked import Choked

choke = Choked(redis_url="redis://localhost:6379/0")

@choke(key="shared_resource", request_limit="10/m")
def function_a():
    return "A"

@choke(key="shared_resource", request_limit="10/m")
def function_b():
    return "B"

# Both functions compete for the same 10 requests/minute
```

## Behavior Details

### Automatic Function Detection

* The decorator automatically detects if the wrapped function is async or sync
* No special configuration needed for async functions
* Both sync and async functions can use the same decorator parameters

### Token Estimation

* Token estimators automatically extract text from function arguments
* Supports string arguments, keyword arguments, lists of strings
* Special handling for OpenAI message format: `[{"role": "user", "content": "text"}]`
* Graceful fallback if estimation fails

### Rate Limiting Logic

* **Request-only**: Each call consumes 1 request token
* **Token-only**: Each call consumes estimated tokens based on input
* **Dual limiting**: Each call must acquire both request tokens AND estimated tokens
* Both limits must be satisfied for function to proceed

### Exponential Backoff

* When rate limited, sleep time doubles on each retry
* Random jitter (0.8x to 1.2x) applied to prevent thundering herd
* Automatic retry until tokens become available

### Error Handling

* Network failures result in automatic retry with backoff
* Token estimation failures fall back to simpler estimators
* Invalid rate formats raise `ValueError` immediately

## Validation

### Constructor Validation

```python theme={null}
# Must specify exactly one backend
Choked()  # ValueError: Must specify either redis_url or api_token
Choked(redis_url="...", api_token="...")  # ValueError: Cannot specify both

# Valid constructors
Choked(redis_url="redis://localhost:6379/0")  # ✓
Choked(api_token="your-token")  # ✓
```

### Decorator Validation

```python theme={null}
choke = Choked(redis_url="redis://localhost:6379/0")

# Must provide at least one limit
@choke(key="api")  # ValueError: At least one limit must be provided

# Invalid rate formats
@choke(key="api", request_limit="invalid")  # ValueError: Invalid rate format
@choke(key="api", request_limit="10/h")     # ValueError: Invalid period 'h'
@choke(key="api", request_limit="10")       # ValueError: Missing period

# Valid decorators
@choke(key="api", request_limit="10/s")  # ✓
@choke(key="api", token_limit="1000/m", token_estimator="openai")  # ✓
@choke(key="api", request_limit="10/s", token_limit="1000/m", token_estimator="openai")  # ✓
```

## Performance Considerations

### Backend Performance

* **Redis backend**: Atomic Lua scripts, minimal network overhead
* **Managed service**: HTTP-based, additional network latency
* Both backends are optimized for high-throughput scenarios

### Token Estimation Performance

* Tokenizers are cached after first use
* Estimation is fast for typical text sizes
* Fallback mechanisms prevent blocking on estimation failures

### Memory Usage

* Minimal memory footprint per rate limit bucket
* Token estimators cache models efficiently
* No memory leaks in long-running applications

## Thread and Process Safety

* All operations are thread-safe and process-safe
* Redis backend uses atomic operations
* Safe for use in multi-threaded web applications
* Supports distributed rate limiting across multiple processes/servers

## Integration Examples

### FastAPI Integration

```python theme={null}
from fastapi import FastAPI
from choked import Choked

app = FastAPI()
choke = Choked(redis_url="redis://localhost:6379/0")

@app.post("/chat")
@choke(key="openai_chat", request_limit="50/s", token_limit="100000/m", token_estimator="openai")
async def chat_endpoint(messages: list[dict]):
    return openai.chat.completions.create(model="gpt-4", messages=messages)
```

### Flask Integration

```python theme={null}
from flask import Flask, request
from choked import Choked

app = Flask(__name__)
choke = Choked(redis_url="redis://localhost:6379/0")

@app.route("/api", methods=["POST"])
@choke(key="api_endpoint", request_limit="100/m")
def api_endpoint():
    return {"result": "success"}
```

### Celery Task Integration

```python theme={null}
from celery import Celery
from choked import Choked

celery_app = Celery("tasks")
choke = Choked(redis_url="redis://localhost:6379/0")

@celery_app.task
@choke(key="background_task", request_limit="10/s")
def background_task(data):
    # Process data with rate limiting
    return process_data(data)
```