pub const PYTHON_SDK_TEMPLATE: &str = r#"# Chasm Python SDK
# Auto-generated - Do not edit directly
# Version: {{version}}
"""
Chasm Python SDK
A Python client library for the Chasm API.
Usage:
from chasm import ChasmClient
client = ChasmClient(api_key="your-api-key")
sessions = client.sessions.list()
"""
import os
import json
import requests
from typing import Optional, List, Dict, Any, Union
from dataclasses import dataclass, field
from datetime import datetime
from urllib.parse import urljoin
__version__ = "{{version}}"
__api_version__ = "{{api_version}}"
@dataclass
class ChasmConfig:
"""Configuration for Chasm client."""
base_url: str = "{{base_url}}"
api_key: Optional[str] = None
timeout: int = 30
retry_count: int = 3
retry_delay: float = 1.0
@dataclass
class Session:
"""Represents a chat session."""
id: str
title: str
provider: str
workspace_id: Optional[str] = None
message_count: int = 0
token_count: int = 0
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None
tags: List[str] = field(default_factory=list)
archived: bool = False
@dataclass
class Message:
"""Represents a chat message."""
id: str
session_id: str
role: str
content: str
model: Optional[str] = None
token_count: int = 0
created_at: Optional[datetime] = None
@dataclass
class Workspace:
"""Represents a workspace."""
id: str
name: str
path: str
provider: str
session_count: int = 0
created_at: Optional[datetime] = None
class ChasmError(Exception):
"""Base exception for Chasm errors."""
def __init__(self, message: str, status_code: Optional[int] = None, response: Optional[dict] = None):
super().__init__(message)
self.status_code = status_code
self.response = response
class AuthenticationError(ChasmError):
"""Authentication failed."""
pass
class RateLimitError(ChasmError):
"""Rate limit exceeded."""
pass
class NotFoundError(ChasmError):
"""Resource not found."""
pass
class ApiClient:
"""Low-level API client."""
def __init__(self, config: ChasmConfig):
self.config = config
self.session = requests.Session()
if config.api_key:
self.session.headers["Authorization"] = f"Bearer {config.api_key}"
self.session.headers["Content-Type"] = "application/json"
self.session.headers["User-Agent"] = f"chasm-python/{__version__}"
def request(self, method: str, path: str, **kwargs) -> dict:
"""Make an API request."""
url = urljoin(self.config.base_url, path)
kwargs.setdefault("timeout", self.config.timeout)
response = self.session.request(method, url, **kwargs)
if response.status_code == 401:
raise AuthenticationError("Invalid API key", 401)
elif response.status_code == 404:
raise NotFoundError("Resource not found", 404)
elif response.status_code == 429:
raise RateLimitError("Rate limit exceeded", 429)
elif response.status_code >= 400:
raise ChasmError(f"API error: {response.text}", response.status_code)
if response.content:
return response.json()
return {}
def get(self, path: str, params: Optional[dict] = None) -> dict:
return self.request("GET", path, params=params)
def post(self, path: str, data: Optional[dict] = None) -> dict:
return self.request("POST", path, json=data)
def put(self, path: str, data: Optional[dict] = None) -> dict:
return self.request("PUT", path, json=data)
def delete(self, path: str) -> dict:
return self.request("DELETE", path)
class SessionsResource:
"""Sessions API resource."""
def __init__(self, client: ApiClient):
self._client = client
def list(
self,
workspace_id: Optional[str] = None,
provider: Optional[str] = None,
archived: Optional[bool] = None,
limit: int = 20,
offset: int = 0,
) -> List[Session]:
"""List sessions."""
params = {"limit": limit, "offset": offset}
if workspace_id:
params["workspace_id"] = workspace_id
if provider:
params["provider"] = provider
if archived is not None:
params["archived"] = str(archived).lower()
response = self._client.get("/api/sessions", params)
return [self._parse_session(s) for s in response.get("sessions", [])]
def get(self, session_id: str) -> Session:
"""Get a session by ID."""
response = self._client.get(f"/api/sessions/{session_id}")
return self._parse_session(response)
def create(self, title: str, provider: str, workspace_id: Optional[str] = None) -> Session:
"""Create a new session."""
data = {"title": title, "provider": provider}
if workspace_id:
data["workspace_id"] = workspace_id
response = self._client.post("/api/sessions", data)
return self._parse_session(response)
def update(self, session_id: str, **kwargs) -> Session:
"""Update a session."""
response = self._client.put(f"/api/sessions/{session_id}", kwargs)
return self._parse_session(response)
def delete(self, session_id: str) -> bool:
"""Delete a session."""
self._client.delete(f"/api/sessions/{session_id}")
return True
def archive(self, session_id: str) -> Session:
"""Archive a session."""
return self.update(session_id, archived=True)
def search(self, query: str, limit: int = 20) -> List[Session]:
"""Search sessions."""
response = self._client.get("/api/sessions/search", {"q": query, "limit": limit})
return [self._parse_session(s) for s in response.get("sessions", [])]
def _parse_session(self, data: dict) -> Session:
return Session(
id=data["id"],
title=data.get("title", "Untitled"),
provider=data.get("provider", "unknown"),
workspace_id=data.get("workspace_id"),
message_count=data.get("message_count", 0),
token_count=data.get("token_count", 0),
tags=data.get("tags", []),
archived=data.get("archived", False),
)
class WorkspacesResource:
"""Workspaces API resource."""
def __init__(self, client: ApiClient):
self._client = client
def list(self, limit: int = 20, offset: int = 0) -> List[Workspace]:
"""List workspaces."""
response = self._client.get("/api/workspaces", {"limit": limit, "offset": offset})
return [self._parse_workspace(w) for w in response.get("workspaces", [])]
def get(self, workspace_id: str) -> Workspace:
"""Get a workspace by ID."""
response = self._client.get(f"/api/workspaces/{workspace_id}")
return self._parse_workspace(response)
def _parse_workspace(self, data: dict) -> Workspace:
return Workspace(
id=data["id"],
name=data.get("name", ""),
path=data.get("path", ""),
provider=data.get("provider", ""),
session_count=data.get("session_count", 0),
)
class HarvestResource:
"""Harvest API resource."""
def __init__(self, client: ApiClient):
self._client = client
def run(self, providers: Optional[List[str]] = None) -> dict:
"""Run harvest."""
data = {}
if providers:
data["providers"] = providers
return self._client.post("/api/harvest", data)
def status(self) -> dict:
"""Get harvest status."""
return self._client.get("/api/harvest/status")
class ChasmClient:
"""Main Chasm client."""
def __init__(
self,
api_key: Optional[str] = None,
base_url: Optional[str] = None,
**kwargs
):
api_key = api_key or os.environ.get("CHASM_API_KEY")
config = ChasmConfig(
api_key=api_key,
base_url=base_url or "{{base_url}}",
**kwargs
)
self._api = ApiClient(config)
# Resources
self.sessions = SessionsResource(self._api)
self.workspaces = WorkspacesResource(self._api)
self.harvest = HarvestResource(self._api)
def health(self) -> dict:
"""Check API health."""
return self._api.get("/health")
def stats(self) -> dict:
"""Get statistics."""
return self._api.get("/api/stats")
# Convenience function
def create_client(**kwargs) -> ChasmClient:
"""Create a Chasm client with environment configuration."""
return ChasmClient(**kwargs)
"#;Expand description
Python SDK template