"""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 (
Quota,
QuotaAlert,
QuotaAlertsPostRequest,
QuotasCheckRequest,
QuotasCheckResponse,
QuotasDeleteResponse,
QuotasPostRequest,
QuotasPutRequest
)
from limitry.client.utils.pagination import PaginatedResponse
if TYPE_CHECKING:
from limitry.client.client import Client
[docs]
class Quotas:
"""Operations for quotas."""
[docs]
def __init__(self, client: "Client") -> None:
"""Initialize Quotas operations.
Args:
client: The Limitry client instance
"""
self._client = client
[docs]
async def check(
self,
request: QuotasCheckRequest
) -> QuotasCheckResponse:
"""Check quotas.
Check usage against all matching quotas for the given dimensions without consuming quota.
This endpoint allows you to query quota status without actually recording usage. Useful for:
- Pre-flight checks before processing requests
- Displaying quota status to users
- Monitoring quota consumption
**Dimension Matching:**
Quotas 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: QuotasCheckRequest object
Returns:
QuotasCheckResponse: Quota check result
Example::
request = QuotasCheckRequest(
# Add request properties here
)
result = await client.quotas.check(
request
)
print(result)
Raises:
AuthenticationError: Unauthorized
NetworkError: If a network error occurs"""
response = await self._client.request("POST", "/quotas/check", json=request.model_dump(mode='json', exclude_none=True))
return QuotasCheckResponse(**response)
[docs]
async def list(
self,
limit: Optional[str] = None,
cursor: Optional[str] = None
) -> PaginatedResponse[Quota]:
"""List quotas.
List all quotas for the project.
Args:
limit: Maximum number of quotas to return (1-100, default: 50) (default: "50")
cursor: Pagination cursor from the previous response
Returns:
PaginatedResponse[Quota]: List of quotas
Example::
items = await client.quotas.list()
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", "/quotas", params=params)
return PaginatedResponse(
data=[Quota(**item) for item in response['data']],
next_cursor=response.get('nextCursor'),
has_more=response.get('hasMore', False)
)
[docs]
async def create(
self,
request: QuotasPostRequest
) -> Quota:
"""Create a quota.
Create a new usage quota with dimension filters.
Quotas allow you to limit usage based on metrics like total tokens, cost, or event count. You can apply quotas to specific customers, models, event types, or any combination using dimension filters.
Args:
request: QuotasPostRequest object
Returns:
Quota: Quota created
Example::
request = QuotasPostRequest(
name="example",
metric="example",
limitValue=123,
dimensionFilters={}
# ... other properties
)
result = await client.quotas.create(
request
)
print(result)
Raises:
APIError: Invalid request
AuthenticationError: Unauthorized
NetworkError: If a network error occurs"""
response = await self._client.request("POST", "/quotas", json=request.model_dump(mode='json', exclude_none=True))
return Quota(**response)
[docs]
async def get(
self,
id: str
) -> Quota:
"""Get a quota.
Get a specific quota by ID.
Args:
id: Unique identifier for the quota
Returns:
Quota: Quota found
Example::
result = await client.quotas.get(
"id_123"
)
print(result)
Raises:
AuthenticationError: Unauthorized
APIError: Quota not found
NetworkError: If a network error occurs"""
response = await self._client.request("GET", f"/quotas/{id}")
return Quota(**response)
[docs]
async def update(
self,
id: str,
request: QuotasPutRequest
) -> Quota:
"""Update a quota.
Update an existing quota.
Args:
id: Unique identifier for the quota
request: QuotasPutRequest object
Returns:
Quota: Quota updated
Example::
request = QuotasPutRequest(
name="example",
dimensionFilters={}
# ... other properties
)
result = await client.quotas.update(
"id_123",
request
)
print(result)
Raises:
AuthenticationError: Unauthorized
APIError: Quota not found
NetworkError: If a network error occurs"""
response = await self._client.request("PUT", f"/quotas/{id}", json=request.model_dump(mode='json', exclude_none=True))
return Quota(**response)
[docs]
async def delete(
self,
id: str
) -> QuotasDeleteResponse:
"""Delete a quota.
Delete a quota by ID.
Args:
id: Unique identifier for the quota
Returns:
QuotasDeleteResponse: Quota deleted
Example::
result = await client.quotas.delete(
"id_123"
)
print(result)
Raises:
AuthenticationError: Unauthorized
APIError: Quota not found
NetworkError: If a network error occurs"""
response = await self._client.request("DELETE", f"/quotas/{id}")
return QuotasDeleteResponse(**response)
[docs]
async def list_alerts(
self,
quota_id: str,
limit: Optional[str] = None,
cursor: Optional[str] = None
) -> PaginatedResponse[QuotaAlert]:
"""List quota alerts.
Get all alerts configured for a quota.
Args:
quota_id: No description provided
limit: Maximum number of alerts to return (1-100, default: 50) (default: "50")
cursor: Pagination cursor from the previous response
Returns:
PaginatedResponse[QuotaAlert]: List of alerts
Example::
items = await client.quotas.list_alerts(
"quotaid_123"
)
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", f"/quotas/{quota_id}/alerts", params=params)
return PaginatedResponse(
data=[QuotaAlert(**item) for item in response['data']],
next_cursor=response.get('nextCursor'),
has_more=response.get('hasMore', False)
)
[docs]
async def create_alert(
self,
quota_id: str,
request: QuotaAlertsPostRequest
) -> QuotaAlert:
"""Create a quota alert.
Configure an alert for when a quota reaches a threshold. Alerts are sent to all active project webhooks.
Args:
quota_id: No description provided
request: QuotaAlertsPostRequest object
Returns:
QuotaAlert: Alert created successfully
Example::
request = QuotaAlertsPostRequest(
threshold=123
)
result = await client.quotas.create_alert(
"quotaid_123",
request
)
print(result)
Raises:
APIError: Invalid request body
AuthenticationError: Unauthorized
NetworkError: If a network error occurs"""
response = await self._client.request("POST", f"/quotas/{quota_id}/alerts", json=request.model_dump(mode='json', exclude_none=True))
return QuotaAlert(**response)