The Python SDK
A small, typed client for CAVS Hub. Streamed hashing, presigned transfers, dedup-aware uploads, and a typed error hierarchy with automatic retries — all behind CAVS.from_env().
Requires Python 3.9 or newer. The SDK sends its version in User-Agent and X-CAVS-Integration-Version, and defaults X-CAVS-Integration to python-sdk.
From install to CI.
Five steps take you from an empty environment to an artifact uploaded and restored inside a pipeline.
1 Install the package
Add cavs to your project. Pin it in requirements.txt or your lockfile for reproducible builds.
$ pip install cavs
2 Create a service account & export a token
In the CAVS dashboard, create a service account scoped to artifacts:read and artifacts:write, then export its key. Service-account keys start with cavs_sk_ and never expire on their own — rotate them, and never commit them.
$ export CAVS_TOKEN=cavs_sk_... $ export CAVS_API=https://cavsnode.com # or your self-hosted Hub
3 Your first upload
upload() streams a sha256 for every file, opens a session, skips objects the Hub already has, PUTs the rest to presigned URLs, then finalizes a new artifact version and returns dedup stats.
from cavs import CAVS client = CAVS.from_env() artifact = client.artifacts.upload( path="./model", project="vision-models", name="resnet50", kind="model", version="1.4.0", ) print(artifact.reference, artifact.deduplication_ratio)
4 Your first download
download() authorizes the object ids, streams each to a temp file while hashing, verifies sha256 == oid (raising ChecksumError on mismatch), then atomically renames into place.
client.artifacts.download( "cavs://acme-ai/vision-models/model/resnet50:1.4.0", dest="./restored", )
5 Use it in CI
Store the token as a secret and call the SDK from any job. The example below publishes a build artifact keyed by the commit sha.
# .github/workflows/publish.yml (job step) - run: pip install cavs && python publish.py env: CAVS_TOKEN: ${{ secrets.CAVS_TOKEN }} CAVS_API: https://cavsnode.com
Typed failures, safe retries.
Every failure raises a subclass of CAVSError, so you can handle exactly the case you care about. Transient failures (429, 5xx, transport) are retried automatically — up to 4 attempts with exponential backoff and jitter, honouring Retry-After (capped ~30s).
- AuthenticationError — 401, bad or missing token
- AuthorizationError — 403, missing scope
- NotFoundError — 404 · ConflictError — 409
- QuotaExceededError — 413 / 402
- RateLimitError — 429 (respects Retry-After)
- ChecksumError — download hash mismatch
from cavs import CAVS from cavs.errors import ( AuthenticationError, QuotaExceededError, ChecksumError, ) client = CAVS.from_env() try: client.artifacts.upload(path="./model", project="p", name="m", kind="model", version="1.0.0") except QuotaExceededError: alert("storage quota hit — clean up old versions") except AuthenticationError: alert("CAVS_TOKEN is invalid or expired")
repr(), exceptions, or logs — it is shown as cavs_sk_…abcd.