SDK¶
High-level SDK for integrating Limitry usage tracking.
Limitry SDK
High-level SDK for integrating Limitry usage tracking and rate limiting.
- class limitry.sdk.Client[source]¶
Bases:
objectMain client for interacting with the Limitry API
- __init__(api_key, base_url=None, timeout=None, headers=None)[source]¶
Initialize the Limitry client.
- Parameters:
Example:
client = Client(api_key="your-api-key")
- async request(method, path, params=None, json=None, headers=None)[source]¶
Make an HTTP request to the API.
- Parameters:
- Return type:
- Returns:
Response data (parsed JSON)
- Raises:
AuthenticationError – When authentication fails (401/403)
APIError – When API request fails
NetworkError – When network error occurs
- class limitry.sdk.ClientConfig[source]¶
Bases:
objectConfiguration for the Limitry SDK client
- exception limitry.sdk.LimitryError[source]¶
Bases:
ExceptionBase exception for all Limitry SDK errors
- exception limitry.sdk.APIError[source]¶
Bases:
LimitryErrorException raised when an API request fails
- exception limitry.sdk.AuthenticationError[source]¶
Bases:
APIErrorException raised when authentication fails (401/403)
- exception limitry.sdk.NetworkError[source]¶
Bases:
LimitryErrorException raised when a network error occurs
- class limitry.sdk.PaginatedResponse[source]¶
Bases:
Generic[T]Response from a paginated API endpoint
- async limitry.sdk.paginate_all(fetch_page, initial_cursor=None)[source]¶
Iterator helper for auto-pagination.
Automatically fetches all pages using cursor-based pagination.
- Parameters:
- Yields:
Individual items from all pages
- Return type:
AsyncGenerator[TypeVar(T),None]
Example:
async def fetch_events(cursor: Optional[str] = None): response = await client.request("GET", "/events", params={"cursor": cursor}) return PaginatedResponse( data=response["data"], next_cursor=response.get("nextCursor"), has_more=response.get("hasMore", False), ) async for event in paginate_all(fetch_events): print(event.id)
- async limitry.sdk.collect_all(fetch_page, initial_cursor=None)[source]¶
Collect all items from paginated endpoint into a list.
- Parameters:
- Return type:
- Returns:
List of all items from all pages
Example:
async def fetch_events(cursor: Optional[str] = None): response = await client.request("GET", "/events", params={"cursor": cursor}) return PaginatedResponse( data=response["data"], next_cursor=response.get("nextCursor"), has_more=response.get("hasMore", False), ) all_events = await collect_all(fetch_events)