tessl/pypi-aleph-alpha-client
Python client to interact with Aleph Alpha API endpoints
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
client-management.mddocs/
Client Management
Core client classes for synchronous and asynchronous API access to Aleph Alpha's language model services. Provides connection management, authentication, retry logic, and comprehensive error handling.
Capabilities
Synchronous Client
The main client class for synchronous API operations with built-in retry logic and connection pooling.
class Client:
def __init__(
self,
token: str,
host: str = "https://api.aleph-alpha.com",
hosting: Optional[str] = None,
request_timeout_seconds: int = 305,
total_retries: int = 8,
nice: bool = False,
verify_ssl: bool = True,
tags: Optional[Sequence[str]] = None,
pool_size: int = 10
):
"""
Initialize synchronous Aleph Alpha client.
Parameters:
- token: API authentication token
- host: API endpoint URL
- hosting: Datacenter preference ("aleph-alpha" or None)
- request_timeout_seconds: Timeout for individual requests
- total_retries: Number of retry attempts for failed requests
- nice: Deprioritize requests for background processing
- verify_ssl: Enable SSL certificate verification
- tags: Internal tagging for request tracking
- pool_size: Connection pool size for HTTP requests
"""
def validate_version(self) -> None:
"""Validate API version compatibility."""
def get_version(self) -> str:
"""Get current API version."""
def models(self) -> List[Mapping[str, Any]]:
"""List available models and their capabilities."""
def tokenizer(self, model: str) -> Tokenizer:
"""Get tokenizer instance for specified model."""Asynchronous Client
Asynchronous client implementation with context manager support and enhanced streaming capabilities.
class AsyncClient:
def __init__(
self,
token: str,
host: str = "https://api.aleph-alpha.com",
hosting: Optional[str] = None,
request_timeout_seconds: int = 305,
total_retries: int = 8,
nice: bool = False,
verify_ssl: bool = True,
tags: Optional[Sequence[str]] = None
):
"""
Initialize asynchronous Aleph Alpha client.
Parameters: Same as Client except pool_size (managed internally)
"""
async def close(self) -> None:
"""Close client session and cleanup resources."""
async def validate_version(self) -> None:
"""Validate API version compatibility (async)."""
async def get_version(self) -> str:
"""Get current API version (async)."""
async def models(self) -> List[Mapping[str, Any]]:
"""List available models and their capabilities (async)."""
async def tokenizer(self, model: str) -> Tokenizer:
"""Get tokenizer instance for specified model (async)."""Error Handling
Custom exception classes for handling API-specific error conditions.
class QuotaError(Exception):
"""Raised when API quota is exceeded (HTTP 402)."""
class BusyError(Exception):
"""Raised when service is temporarily unavailable (HTTP 503)."""Usage Examples
Basic client initialization and usage patterns:
from aleph_alpha_client import Client, AsyncClient, QuotaError, BusyError
# Synchronous client
client = Client(token="your-api-token")
try:
# Check available models
models = client.models()
print(f"Available models: {[m['name'] for m in models]}")
# Validate API compatibility
client.validate_version()
except QuotaError:
print("API quota exceeded")
except BusyError:
print("Service temporarily unavailable")
# Asynchronous client with context manager
import asyncio
async def main():
async with AsyncClient(token="your-api-token") as client:
try:
models = await client.models()
print(f"Available models: {[m['name'] for m in models]}")
except QuotaError:
print("API quota exceeded")
except BusyError:
print("Service temporarily unavailable")
asyncio.run(main())
# Client with custom configuration
client = Client(
token="your-api-token",
host="https://api.aleph-alpha.com",
request_timeout_seconds=120, # 2 minute timeout
total_retries=5, # Fewer retries
nice=True, # Background processing priority
verify_ssl=True # SSL verification enabled
)Install with Tessl CLI
npx tessl i tessl/pypi-aleph-alpha-client