Skip to main content

​
The Check-Record Pattern

Always check limits before, record events after: 
from limitry import Limitry
from openai import OpenAI

limitry = Limitry()
openai = OpenAI()

def chat(customer_id: str, message: str) -> str:
    # 1. Check limits
    check = limitry.limits.check(customer_id=customer_id)
    
    if not check.allowed:
        raise Exception(f"Limit exceeded: {check.limits[0].name}")
    
    # 2. Make request
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": message}]
    )
    
    # 3. Record event
    limitry.events.record(
        customer_id=customer_id,
        event_type="llm.completion",
        values={"tokens": response.usage.total_tokens},
        dimensions={"model": "gpt-4"}
    )
    
    return response.choices[0].message.content

​
Record Events

Log usage after each operation: 
result = limitry.events.record(
    customer_id="cust_123",
    event_type="llm.completion",
    
    # Numeric values (for metering)
    values={
        "tokens": 500,
        "cost_cents": 25,
        "latency_ms": 1200
    },
    
    # Categorical dimensions (for filtering)
    dimensions={
        "model": "gpt-4",
        "provider": "openai",
        "feature": "chat"
    }
)

print(f"Recorded event: {result.id}")

​
Values vs Dimensions

  • Values: Numbers that can be summed, averaged, or maxed by meters
  • Dimensions: Strings for filtering and grouping
# Values are numeric
values={
    "tokens": 500,
    "cost_cents": 25,
    "images": 4
}

# Dimensions are strings
dimensions={
    "model": "gpt-4",
    "team": "engineering",
    "environment": "production"
}

​
List Events

Retrieve recorded events with filtering: 
# List all events
events = limitry.events.list()

for event in events.data:
    print(f"{event.id}: {event.customer_id} - {event.event_type}")
    print(f"  Values: {event.values}")
    print(f"  Dimensions: {event.dimensions}")

# Filter by customer
events = limitry.events.list(customer_id="cust_123")

# Filter by event type
events = limitry.events.list(event_type="llm.completion")

# Filter by date range
events = limitry.events.list(
    start_date="2024-01-01T00:00:00Z",
    end_date="2024-01-31T23:59:59Z"
)

# Pagination
events = limitry.events.list(limit=50)
if events.has_more:
    next_page = limitry.events.list(cursor=events.next_cursor)

​
Get Usage Summary

Aggregate usage over a time range: 
from datetime import datetime, timedelta

summary = limitry.events.get_summary(
    start_date=(datetime.now() - timedelta(days=7)).isoformat(),
    end_date=datetime.now().isoformat(),
    customer_id="cust_123"  # Optional filter
)

print(f"Total events: {summary.total_events}")
print(f"Values: {summary.values}")  # Aggregated values

​
Get Usage Breakdown

Group usage by dimension: 
breakdown = limitry.events.get_breakdown(
    group_by="model",
    start_date="2024-01-01T00:00:00Z",
    end_date="2024-01-31T23:59:59Z"
)

for item in breakdown.data:
    print(f"{item.dimension}: {item.total_events} events")
    print(f"  Values: {item.values}")

​
Batch Recording

Record multiple events efficiently: 
result = limitry.events.batch_ingest(
    events=[
        {
            "customer_id": "cust_123",
            "event_type": "llm.completion",
            "values": {"tokens": 500},
            "dimensions": {"model": "gpt-4"}
        },
        {
            "customer_id": "cust_456",
            "event_type": "llm.embedding",
            "values": {"tokens": 1000},
            "dimensions": {"model": "text-embedding-3-small"}
        }
    ]
)

print(f"Recorded {result.count} events")

​
Event Types

Use descriptive event types: 
# LLM operations
event_type="llm.completion"
event_type="llm.embedding"

# Image operations
event_type="image.generation"

# Agent operations
event_type="agent.run"

# Resource tracking
event_type="agent.count"

# Custom operations
event_type="feature.usage"

​
Example: Full Integration

from limitry import Limitry
from openai import OpenAI

limitry = Limitry()
openai = OpenAI()

def process_request(customer_id: str, prompt: str) -> str:
    # 1. Check limits
    check = limitry.limits.check(customer_id=customer_id)
    if not check.allowed:
        exceeded = [l for l in check.limits if l.exceeded]
        raise Exception(f"Limit exceeded: {exceeded[0].name}")
    
    # 2. Make the LLM call
    start_time = time.time()
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )
    latency_ms = int((time.time() - start_time) * 1000)
    
    # 3. Record comprehensive event
    limitry.events.record(
        customer_id=customer_id,
        event_type="llm.completion",
        values={
            "input_tokens": response.usage.prompt_tokens,
            "output_tokens": response.usage.completion_tokens,
            "total_tokens": response.usage.total_tokens,
            "latency_ms": latency_ms
        },
        dimensions={
            "model": "gpt-4",
            "provider": "openai",
            "feature": "chat"
        }
    )
    
    return response.choices[0].message.content