API Reference

from chamber_sdk import ChamberClient

# From environment variable (CHAMBER_TOKEN)
client = ChamberClient()

# From CLI config (~/.chamber/token.json)
client = ChamberClient.from_config()

# Direct token
client = ChamberClient(
    token="ch.your-api-token",
    organization_id="org-123",  # Optional: for multi-org users
    api_url="https://custom.api.example.com/v1",  # Optional: override endpoint
    timeout=60  # Optional: request timeout in seconds
)

# Configure registries once
ChamberClient.add_registry("prod", "us-east1-docker.pkg.dev/my-project/prod", set_default=True)
ChamberClient.add_registry("dev", "us-east1-docker.pkg.dev/my-project/dev")

# Submit using default registry
job = client.run("./my-project", gpus=4, team="ml-research")

# Submit to specific registry by name
job = client.run("./my-project", gpus=4, team="ml-research", registry="dev")

# Full URLs still work
job = client.run("./my-project", registry="123456.dkr.ecr.us-east-1.amazonaws.com", team="ml-research")

# Dry run (preview without executing)
result = client.run("./my-project", dry_run=True)
print(result.dockerfile)
print(result.manifest)

registries = ChamberClient.list_registries()
# {'prod': 'us-east1-docker.pkg.dev/my-project/prod', 'dev': '...'}

ChamberClient.add_registry(
    "prod",
    "us-east1-docker.pkg.dev/my-project/prod",
    set_default=True  # Optional: also set as default
)

ChamberClient.set_default_registry("dev")

name, url = ChamberClient.get_default_registry()
# ('prod', 'us-east1-docker.pkg.dev/my-project/prod')

job = client.submit_job(
    name="training-job",
    initiative_id="team-id",
    gpu_type="H100",
    requested_gpus=8,
    job_class=JobClass.RESERVED,
    priority=50,
    tags={"experiment": "v1", "owner": "ml-team"},
    external_id="my-tracking-id"
)

job = client.submit_job(
    name="distributed-training",
    initiative_id="team-id",
    gpu_type="H100",
    requested_gpus=32,
    gpus_per_pod=8,
    requested_pods=4,
    scaling_mode=ScalingMode.GANG,  # All-or-nothing scheduling
    distribution_mode="auto"
)

job = client.get_workload("workload-id")
print(f"Status: {job.status}")
print(f"GPUs: {job.requested_gpus}")

response = client.list_workloads(
    status=JobStatus.RUNNING,
    initiative_id="team-id",
    submitted_by="user-123",
    limit=50
)

for job in response.items:
    print(f"{job.name}: {job.status}")

for job in client.iter_workloads(status=JobStatus.COMPLETED):
    print(f"{job.name}: {job.requested_gpus} GPUs")

cancelled_job = client.cancel_workload("workload-id")
print(f"Status: {cancelled_job.status}")  # CANCELLED

results = client.search_workloads(
    status=["RUNNING", "PENDING"],
    gpu_type=["H100"],
    priority_min=50,
    submitted_from="2024-01-01T00:00:00Z",
    query="training",
    sort_by="submitted_at",
    sort_order="desc",
    page_size=25
)

print(f"Found {results.total_count} workloads")
for job in results.items:
    print(f"{job.name}: {job.status}")

# Paginate through results
if results.has_more:
    next_page = client.search_workloads(cursor=results.next_cursor)

agg = client.get_workload_aggregations(
    dimension="status",
    initiative_id=["team-ml"]
)

print(f"Total: {agg.total}")
for bucket in agg.buckets:
    print(f"  {bucket.key}: {bucket.count}")

result = client.wait_for_completion(
    workload_id="job-id",
    poll_interval=30,  # Seconds between checks
    timeout=3600       # Max wait time in seconds
)

metrics = client.get_workload_metrics(
    "workload-id",
    time_range="job_lifetime"
)

print(f"GPU Utilization: {metrics.gpu_utilization.avg:.1f}%")
print(f"Memory: {metrics.memory_utilization.avg:.1f}%")
print(f"Temperature: {metrics.temperature.avg:.1f}C")
print(f"Power: {metrics.power_usage.avg:.1f} W")

metrics = client.get_global_metrics(
    time_range="last_24h",
    initiative_id="team-ml"
)

print(f"Active workloads: {metrics.active_workloads}")
print(f"Total GPU hours: {metrics.total_gpu_hours}")
print(f"Avg GPU utilization: {metrics.gpu_utilization.avg:.1f}%")

data = client.get_batch_workload_metrics(
    status=["RUNNING"],
    time_range="last_24h",
    sort_by="gpu_utilization",
    limit=10
)

stats = client.get_workload_stats(
    time_range="last_7_days",
    initiative_id="team-id"
)

print(f"Total workloads: {stats.total}")
for status, count in stats.by_status.items():
    print(f"  {status}: {count}")

teams = client.list_teams()

for team in teams:
    print(f"{team.name} ({team.id})")

team = client.create_team(
    name="ML Research",
    description="Machine learning research team",
    tags={"department": "engineering"}
)

print(f"Created team: {team.id}")

team = client.get_team("team-id")
print(f"Team: {team.name}")
print(f"Description: {team.description}")

templates = client.list_templates(
    scope="ORGANIZATION",
    include_system=True
)

for template in templates:
    print(f"{template.name}: {template.gpu_type}")

template = client.get_template("template-id")
print(f"Template: {template.name}")
print(f"GPU Type: {template.gpu_type}")
print(f"GPUs: {template.requested_gpus}")

allocations = client.list_allocations(initiative_id="team-id")

for alloc in allocations:
    print(f"{alloc.id}: {alloc.allocated_instances} instances ({alloc.status})")

allocation = client.create_allocation(
    initiative_id="team-id",
    reservation_id="reservation-123",
    allocated_instances=4
)

print(f"Allocated: {allocation.id}")

allocation = client.get_allocation("allocation-id")
print(f"Status: {allocation.status}")
print(f"Instances: {allocation.allocated_instances}")

capacity = client.get_capacity()

# Budget info
print(f"Allocated: {capacity.budget.allocated} GPU-hours")
print(f"Used: {capacity.budget.used} GPU-hours")
print(f"Available: {capacity.budget.available} GPU-hours")

# Capacity pools
for pool in capacity.pools:
    print(f"{pool.name} ({pool.gpu_type}):")
    print(f"  Available: {pool.available_gpus}/{pool.total_gpus} GPUs")

health = client.health()
print(f"Status: {health.status}")
print(f"Version: {health.version}")

from chamber_sdk import JobStatus

JobStatus.PENDING      # Submitted, awaiting scheduling
JobStatus.QUEUED       # Scheduled, waiting for resources
JobStatus.STARTING     # Resources allocated, starting
JobStatus.RUNNING      # Currently executing
JobStatus.COMPLETED    # Finished successfully
JobStatus.FAILED       # Workload failed
JobStatus.PREEMPTED    # Preempted (elastic jobs)
JobStatus.CANCELLED    # Cancelled by user

from chamber_sdk import JobClass

JobClass.RESERVED   # Guaranteed capacity, non-preemptible
JobClass.ELASTIC    # Uses idle capacity, can be preempted
JobClass.DISCOVERED # External workload discovered by Chamber

from chamber_sdk import ScalingMode

ScalingMode.GANG     # All-or-nothing scheduling
ScalingMode.ELASTIC  # Can scale between min/max pods

from chamber_sdk import AllocationStatus

AllocationStatus.ACTIVE    # Currently active
AllocationStatus.RELEASED  # Released by user
AllocationStatus.EXPIRED   # Expired

from chamber_sdk import (
    ChamberError,
    AuthenticationError,
    AuthorizationError,
    NotFoundError,
    ValidationError,
    RateLimitError,
    ServerError,
    DockerError,  # For run() errors
)
from chamber_sdk.run import RegistryAuthError  # Specific registry auth failures

try:
    job = client.get_workload("invalid-id")
except NotFoundError:
    print("Workload not found")
except AuthenticationError:
    print("Invalid or expired token")
except RateLimitError as e:
    print(f"Rate limited: {e.message}")
except ChamberError as e:
    print(f"API error: {e}")

from chamber_sdk import ChamberClient, DockerError
from chamber_sdk.run import RegistryAuthError

try:
    job = client.run("./my-project", registry="...", team="...")
except RegistryAuthError as e:
    # Registry authentication failed (401/403 from registry)
    # e.registry contains the registry URL
    # e.detail contains the error details
    print(f"Registry auth failed for {e.registry}: {e.detail}")
except DockerError as e:
    # Docker not installed, daemon not running, build failed, or push failed
    print(f"Docker error: {e}")
except FileNotFoundError as e:
    # No Python entrypoint found in project
    print(f"Project error: {e}")
except ValueError as e:
    # Missing required parameters (registry, team)
    print(f"Configuration error: {e}")