"""Auto-generated operation classes.
This file is auto-generated by the operators generator.
Do not edit this file manually.
"""
from typing import TYPE_CHECKING, Any, Dict, List, Optional
from limitry.client.types import (
RateLimit,
RateLimitsCheckRequest,
RateLimitsCheckResponse,
RateLimitsDeleteResponse,
RateLimitsPostRequest,
RateLimitsPutRequest
)
from limitry.client.utils.pagination import PaginatedResponse
if TYPE_CHECKING:
from limitry.client.client import Client
[docs]
class RateLimits:
"""Operations for ratelimits."""
[docs]
def __init__(self, client: "Client") -> None:
"""Initialize RateLimits operations.
Args:
client: The Limitry client instance
"""
self._client = client
[docs]
async def check(
self,
request: RateLimitsCheckRequest
) -> RateLimitsCheckResponse:
"""Check rate limits.
Check current rate limit status for the given dimensions without consuming a request.
This endpoint allows you to query rate limit status without actually counting against the limit. Useful for:
- Pre-flight checks before processing requests
- Displaying rate limit status to users
- Monitoring rate limit consumption
**Dimension Matching:**
Rate limits are matched based on dimension filters. Pass dimensions like:
- `customer_id`: Filter by customer
- `event_type`: Filter by event type
- `model`: Filter by model
- `provider`: Filter by provider
- Any custom dimension from event properties.
Args:
request: RateLimitsCheckRequest object
Returns:
RateLimitsCheckResponse: Rate limit check result
Example::
request = RateLimitsCheckRequest(
# Add request properties here
)
result = await client.rate-limits.check_rate_limits(
request
)
print(result)
Raises:
AuthenticationError: Unauthorized
NetworkError: If a network error occurs"""
response = await self._client.request("POST", "/rate-limits/check", json=request.model_dump(mode='json', exclude_none=True))
return RateLimitsCheckResponse(**response)
[docs]
async def list(
self,
limit: Optional[str] = None,
cursor: Optional[str] = None
) -> PaginatedResponse[RateLimit]:
"""List rate limits.
List all rate limits for the project.
Args:
limit: Maximum number of rate limits to return (1-100, default: 50) (default: "50")
cursor: Pagination cursor from the previous response
Returns:
PaginatedResponse[RateLimit]: List of rate limits
Example::
items = await client.rate-limits.list_rate_limits()
for item in items.data:
print(item)
Raises:
APIError: Invalid query parameters
AuthenticationError: Unauthorized
NetworkError: If a network error occurs"""
params = {
"limit": limit,
"cursor": cursor,
}
params = {k: v for k, v in params.items() if v is not None}
response = await self._client.request("GET", "/rate-limits", params=params)
return PaginatedResponse(
data=[RateLimit(**item) for item in response['data']],
next_cursor=response.get('nextCursor'),
has_more=response.get('hasMore', False)
)
[docs]
async def create(
self,
request: RateLimitsPostRequest
) -> RateLimit:
"""Create a rate limit.
Create a new rate limit with dimension filters.
Rate limits control how many requests can be made within a time window. You can apply rate limits to specific customers, models, event types, or any combination using dimension filters.
Args:
request: RateLimitsPostRequest object
Returns:
RateLimit: Rate limit created
Example::
request = RateLimitsPostRequest(
name="example",
limitValue=123,
window="example",
dimensionFilters={},
enabled=True
)
result = await client.rate-limits.create_rate_limit(
request
)
print(result)
Raises:
APIError: Invalid request
AuthenticationError: Unauthorized
NetworkError: If a network error occurs"""
response = await self._client.request("POST", "/rate-limits", json=request.model_dump(mode='json', exclude_none=True))
return RateLimit(**response)
[docs]
async def get(
self,
id: str
) -> RateLimit:
"""Get a rate limit.
Get a specific rate limit by ID.
Args:
id: Unique identifier for the rate limit
Returns:
RateLimit: Rate limit found
Example::
result = await client.rate-limits.get_rate_limit(
"id_123"
)
print(result)
Raises:
AuthenticationError: Unauthorized
APIError: Rate limit not found
NetworkError: If a network error occurs"""
response = await self._client.request("GET", f"/rate-limits/{id}")
return RateLimit(**response)
[docs]
async def update(
self,
id: str,
request: RateLimitsPutRequest
) -> RateLimit:
"""Update a rate limit.
Update an existing rate limit.
Args:
id: Unique identifier for the rate limit
request: RateLimitsPutRequest object
Returns:
RateLimit: Rate limit updated
Example::
request = RateLimitsPutRequest(
name="example",
dimensionFilters={}
# ... other properties
)
result = await client.rate-limits.update_rate_limit(
"id_123",
request
)
print(result)
Raises:
AuthenticationError: Unauthorized
APIError: Rate limit not found
NetworkError: If a network error occurs"""
response = await self._client.request("PUT", f"/rate-limits/{id}", json=request.model_dump(mode='json', exclude_none=True))
return RateLimit(**response)
[docs]
async def delete(
self,
id: str
) -> RateLimitsDeleteResponse:
"""Delete a rate limit.
Delete a rate limit by ID.
Args:
id: Unique identifier for the rate limit
Returns:
RateLimitsDeleteResponse: Rate limit deleted
Example::
result = await client.rate-limits.delete_rate_limit(
"id_123"
)
print(result)
Raises:
AuthenticationError: Unauthorized
APIError: Rate limit not found
NetworkError: If a network error occurs"""
response = await self._client.request("DELETE", f"/rate-limits/{id}")
return RateLimitsDeleteResponse(**response)