Usage
Sync
Use this page when integrating rsspot into Python services, workers, or automation scripts.
Client Surfaces
| Client | Purpose |
|---|---|
SpotClient | Unified sync + async methods on one object |
AsyncSpotClient | Explicit async-only usage |
Construction Patterns
| Pattern | Use when | Notes |
|---|---|---|
get_client(...) | Process-level reuse for unified client | Singleton cache keyed by (profile, org, region) |
get_async_client(...) | Process-level reuse for async-only client | Singleton cache keyed by (profile, org, region) |
SpotClient(...) | Explicit lifecycle control | Call close() / aclose() |
AsyncSpotClient(...) | Explicit async lifecycle | Use async with or await aclose() |
Unified Sync/Async Pattern
from rsspot import SpotClient
client = SpotClient(profile="prod")
# Sync
orgs = client.organizations.list()
# Async on same object
orgs_async = await client.aorganizations.list()
await client.aclose()
Async-only Pattern
import asyncio
from rsspot import AsyncSpotClient
async def main() -> None:
async with AsyncSpotClient(profile="prod") as client:
regions = await client.regions.list()
print([r.name for r in regions])
asyncio.run(main())
Singleton and Context Helpers
from rsspot import get_client, use_profile, use_region
default_client = get_client(profile="prod")
with use_profile("staging"), use_region("us-east-iad-1"):
scoped = get_client()
print(scoped.regions.list())
Persistent defaults:
set_default_profile(...)set_default_org(...)set_default_region(...)
Auth, Retry, and Cache Behavior
authenticate()reuses valid JWT id tokens by expiry check- On auth failure (
401/403), one forced refresh retry is attempted - Retry policy is configurable via
RetryConfig(attempts, delays, jitter, statuses) - GET requests can be cached via
CacheConfigin sqlite + in-memory front cache
State and Workflow Primitives
StateStore provides persistent state operations for:
- preferences
- HTTP cache
- command history
- registration ledger
RegistrationWorkflow provides idempotent VM registration state transitions:
- discovered
- token issued
- submitted
- registered
- failed
- skipped
Error Handling Pattern
from rsspot.errors import APIError, AuthError, ConfigError, RequestError
try:
client = SpotClient(profile="prod")
events = client.inventory.list_organization_events(limit=50)
client.close()
except ConfigError:
...
except AuthError:
...
except APIError:
...
except RequestError:
...