Clouditia Manager SDK

Python SDK to manage remote GPU sessions on Clouditia via the Computing API (sk_compute_).

Installation

pip install clouditia-manager

This automatically installs the clouditia SDK (for code execution) and boto3 (for S3 uploads).

1. Configuration

from clouditia_manager import GPUManager

# Default configuration (URL: https://clouditia.com/jobs)
manager = GPUManager(api_key="sk_compute_xxxxx")

# Advanced configuration
manager = GPUManager(
    api_key="sk_compute_xxxxx",
    base_url="https://clouditia.com/jobs",  # API URL (default)
    timeout=120                              # Timeout in seconds (default: 60)
)

# Your API key is verified automatically
print(f"User: {manager.user['username']}")
print(f"Email: {manager.user['email']}")

Where to find your API key? Go to clouditia.com/manage/api-keys/ to create a sk_compute_ key.

2. Check available GPUs

Before launching a session, check GPU inventory in real-time:

inventory = manager.get_inventory()

if not inventory:
    print("No GPUs available")
else:
    for gpu in inventory:
        print(f"{gpu.gpu_name} ({gpu.price_per_hour}EUR/h) : "
              f"{gpu.available} available, {gpu.on_demand} on-demand || "
              f"datacenter : {gpu.datacenter},  "
              f"[datacenter_id : {gpu.datacenter_id}]")

Example output:

NVIDIA RTX 3090 (1.0EUR/h) : 1 available, 0 on-demand || datacenter : France-Poissy,  [datacenter_id : e7aabe3c-...]
NVIDIA RTX 3060 Ti (0.5EUR/h) : 0 available, 4 on-demand || datacenter : France-vo8,  [datacenter_id : 487754eb-...]

Status meanings:

• available: GPUs on powered-on machines, ready immediately
• on_demand: GPUs on powered-off machines, startup in ~2-5 minutes
• in_use: GPUs currently used by active sessions

Filter by datacenter

Use the datacenter_id (UUID) to filter inventory for a specific datacenter:

inventory = manager.get_inventory(datacenter_id="487754eb-a676-4502-a0f4-21a88e52c25a")
for gpu in inventory:
    print(f"{gpu.gpu_name}: {gpu.available} available, {gpu.on_demand} on-demand")

List datacenters

datacenters = manager.list_datacenters()

for dc in datacenters:
    print(f"{dc.name} (datacenter_id={dc.datacenter_id}) ,  GPUs: {dc.gpu_count}")

# Example output:
# France-Poissy (datacenter_id=e7aabe3c-...) ,  GPUs: 2
# France-vo8 (datacenter_id=487754eb-...) ,  GPUs: 4

3. Create a GPU session

Once you've chosen a GPU, launch a session. The SDK automatically waits until the session is ready:

session = manager.create_session(
    gpu_type="nvidia-rtx-3090",  # GPU type slug (from inventory)
    vcpu=2,                       # Number of vCPUs
    ram=4,                        # RAM in GB
    storage=20                    # Storage in GB
)

print(f"Session ready: {session.name}")
print(f"VS Code URL: {session.url}")
print(f"Password: {session.password}")

Target a specific datacenter

Use datacenter_id to launch the session on a specific datacenter:

session = manager.create_session(
    gpu_type="nvidia-rtx-3060-ti",
    vcpu=2, ram=4, storage=20,
    datacenter_id="487754eb-a676-4502-a0f4-21a88e52c25a"  # France-vo8
)

Progress tracking

The SDK monitors each creation step (power on, deployment, etc.) and displays progress in real-time. No need to set a timeout — the SDK detects real errors and returns them immediately:

session = manager.create_session(
    gpu_type="nvidia-rtx-3090",
    vcpu=4, ram=16, storage=20,
    wait_ready=True,     # Wait for session to be ready (default: True)
    verbose=True         # Show steps in real-time (default: True)
)

# Typical output (on-demand node):
# Waiting for session f091ef1c to be ready...
#   [powering_on] Powering on node...
#   [waiting_nodes] Waiting for nodes...
#   [deploying] Deploying GPU session...
#   [waiting_ready] Waiting for pod ready...
#
#   SESSION READY
#   ...

# Silent mode (no waiting, no output)
session = manager.create_session(
    gpu_type="nvidia-rtx-3090",
    vcpu=2, ram=4, storage=20,
    wait_ready=False,
    verbose=False
)

4. Stop a session

# Standard stop (waits for pod deletion)
result = manager.stop_session(f"{session.short_id}")  # ex: "0e4c713a"
print(f"Session stopped: {result.name}")

# Silent mode
result = manager.stop_session(f"{session.short_id}", wait_stopped=False, verbose=False)

5. Execute code in a session

To execute code in an active session, generate a sk_live_ key and use the clouditia SDK.

Step 1: Generate an execution key

# Generate a sk_live_ key linked to the session
sdk_key = manager.generate_sdk_key(
    session_id=session.short_id,  # ex: "0e4c713a"
    name="My Execution Key"
)
print(f"Key: {sdk_key}")  # sk_live_xxxxx...

Step 2: Execute shell commands

from clouditia import GPUSession

session_live_gpu = GPUSession(api_key=sdk_key)

# Install system packages
result = session_live_gpu.shell("sudo apt update && sudo apt install -y ffmpeg")
print(result)

# Verify installation
result = session_live_gpu.shell("ffmpeg -version")
print(result)

# Install Python packages
result = session_live_gpu.shell("pip install transformers accelerate")
print(result)

# Verify installation
result = session_live_gpu.shell("python3 -c 'import transformers; print(transformers.__version__)'")
print(result)

# Execute a script from the workspace
result = session_live_gpu.shell("cd /home/coder/workspace && python3 train.py")
print(result)

Step 3: Execute Python code

# Execute Python code directly (via session_live_gpu.run)
result = session_live_gpu.run("""
import torch
print(f'CUDA: {torch.cuda.is_available()}')
print(f'GPU: {torch.cuda.get_device_name(0)}')
a = torch.randn(1000, 1000, device='cuda')
b = torch.randn(1000, 1000, device='cuda')
c = torch.matmul(a, b)
print(f'Result: {c.shape}')
""")
print(result)

Alternative: Lambda GPU (no session management)

For quick execution without creating a session manually, use lambda_gpu() (see section 10).

6. Manage sessions

List sessions

# All sessions
sessions = manager.list_sessions()

# Filter by status
running = manager.list_sessions(status="running")
stopped = manager.list_sessions(status="stopped")

for session in sessions:
    print(f"{session.name} ({session.short_id}): {session.status} - {session.gpu_type}")

Get session details

session = manager.get_session("0e4c713a")

print(f"Name: {session.name}")
print(f"Status: {session.status}")
print(f"GPU: {session.gpu_type}")
print(f"URL: {session.url}")
print(f"Password: {session.password}")

Rename a session

session = manager.rename_session("0e4c713a", "my-ml-project-v1")
print(f"New name: {session.name}")

Session cost and duration

cost_info = manager.get_session_cost("0e4c713a")
print(f"Current cost: {cost_info['cost']} EUR")
print(f"Hourly rate: {cost_info['hourly_rate']} EUR/h")
print(f"Duration: {cost_info['duration_display']}")

Check credit balance

balance = manager.get_balance()
print(f"Balance: {balance['balance']} {balance['currency']}")

7. Multi-GPU sessions

Create a session with multiple GPUs, optionally of different types:

session = manager.create_session(
    gpus=[
        {'type': 'nvidia-rtx-3090', 'count': 1},
        {'type': 'nvidia-rtx-3060-ti', 'count': 1}
    ],
    vcpu=4,
    ram=16,
    storage=20
)

print(f"GPU Count: {session.gpu_count}")
print(f"GPUs: {session.gpus}")

GPU availability (allow_partial, auto_add_gpus)

If some requested GPUs are not available:

# Default: raises an error if any GPU is unavailable
session = manager.create_session(
    gpus=[
        {'type': 'nvidia-rtx-3090', 'count': 1},
        {'type': 'nvidia-rtx-4090', 'count': 1}  # If unavailable -> error
    ],
    vcpu=4, ram=16, storage=20
)
# InsufficientResourcesError: Some GPUs unavailable: nvidia-rtx-4090.
# Use allow_partial=True to create with available GPUs only.

# allow_partial=True: create immediately with available GPUs only
session = manager.create_session(
    gpus=[
        {'type': 'nvidia-rtx-3090', 'count': 1},
        {'type': 'nvidia-rtx-4090', 'count': 1}
    ],
    vcpu=4, ram=16, storage=20,
    allow_partial=True  # Starts with 3090 only, no error
)

# auto_add_gpus=True: start with available GPUs, then automatically
# add missing GPUs when they become available (checks every 30s)
session = manager.create_session(
    gpus=[
        {'type': 'nvidia-rtx-3090', 'count': 1},
        {'type': 'nvidia-rtx-4090', 'count': 1}
    ],
    vcpu=4, ram=16, storage=20,
    allow_partial=True,
    auto_add_gpus=True  # Background thread watches for nvidia-rtx-4090
)

8. Auto-stop limits

Set limits to automatically stop a session:

# Cost limit: auto-stop at 5 EUR
session = manager.create_session(
    gpu_type="nvidia-rtx-3090",
    vcpu=4, ram=16,
    cost_limit=5.0
)

# Duration limit: auto-stop after 2 hours
session = manager.create_session(
    gpu_type="nvidia-rtx-3090",
    vcpu=4, ram=16,
    duration_limit=7200  # seconds
)

# Both: stop when either limit is reached
session = manager.create_session(
    gpu_type="nvidia-rtx-3090",
    vcpu=4, ram=16,
    cost_limit=10.0,
    duration_limit=3600
)

print(f"Auto-stop enabled: {session.auto_stop_enabled}")
print(f"Cost limit: {session.cost_limit} EUR")
print(f"Duration limit: {session.duration_limit} seconds")

9. Queue

If GPUs are not immediately available, place your request in a queue:

result = manager.create_session(
    gpu_type="nvidia-rtx-4090",
    vcpu=4, ram=16, storage=20,
    queue_if_unavailable=True
)

if hasattr(result, 'name'):
    print(f"Session created: {result.name}")
elif isinstance(result, dict) and result.get('queued'):
    print(f"Queued! Position: #{result['position']}")
    print(f"Queue ID: {result['queue_id']}")

Manage the queue

# List queue jobs
queue_jobs = manager.list_queue_jobs()
for job in queue_jobs:
    print(f"Position #{job.position}: {job.status_display}")

# Job details with history
result = manager.get_queue_job("a1b2c3d4", verbose=True)

# Cancel a job
manager.cancel_queue_job("a1b2c3d4")

10. Lambda GPU (Serverless Execution)

Execute Python code directly on a remote GPU, without managing a session:

result = manager.lambda_gpu(
    script="""
import torch
print(f"CUDA: {torch.cuda.is_available()}")
if torch.cuda.is_available():
    print(f"GPU: {torch.cuda.get_device_name(0)}")
    a = torch.randn(5000, 5000, device='cuda')
    b = torch.randn(5000, 5000, device='cuda')
    c = torch.matmul(a, b)
    print(f"5000x5000 matmul OK, shape: {c.shape}")
""",
    gpu_type="nvidia-rtx-3060-ti",
    vcpu=2,
    ram=4,
    storage=20
)

print(f"Exit code: {result.exit_code}")
print(f"Output: {result.stdout}")
print(f"Cost: {result.cost} EUR")
print(f"Duration: {result.duration_seconds}s")

Lambda with custom Docker environment

result = manager.lambda_gpu(
    script="import torch; print(torch.cuda.is_available())",
    gpu_type="nvidia-rtx-3090",
    environment_id="3a07d1e9-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
    vcpu=8, ram=32, storage=100
)

run_and_stop() (Full session with S3 upload)

Creates a session, uploads local files, executes a script on remote GPU, uploads results to S3, then stops the session automatically.

Step 1: Create local files

# Create train.py locally
with open("train.py", "w") as f:
    f.write("""
import torch
import torch.nn as nn
import torch.optim as optim
import pandas as pd
import time

print("=" * 50)
print("  CLOUDITIA GPU TRAINING")
print("=" * 50)

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print(f"Device: {device}")
if torch.cuda.is_available():
    print(f"GPU: {torch.cuda.get_device_name(0)}")

# Load data
df = pd.read_csv('/home/coder/workspace/data.csv')
print(f"Data loaded: {len(df)} rows, {len(df.columns)} columns")

# Simple model
model = nn.Sequential(
    nn.Linear(len(df.columns) - 1, 64),
    nn.ReLU(),
    nn.Linear(64, 32),
    nn.ReLU(),
    nn.Linear(32, 3)
).to(device)

optimizer = optim.Adam(model.parameters(), lr=0.001)
criterion = nn.CrossEntropyLoss()

print(f"Model parameters: {sum(p.numel() for p in model.parameters()):,}")

# Training loop
epochs = 10
for epoch in range(epochs):
    x = torch.randn(64, len(df.columns) - 1).to(device)
    y = torch.randint(0, 3, (64,)).to(device)

    optimizer.zero_grad()
    output = model(x)
    loss = criterion(output, y)
    loss.backward()
    optimizer.step()

    acc = (output.argmax(1) == y).float().mean().item()
    print(f"Epoch {epoch+1}/{epochs} | Loss: {loss.item():.4f} | Acc: {acc:.2%}")

# Save model
torch.save(model.state_dict(), '/home/coder/workspace/model.pt')
print(f"Model saved to /home/coder/workspace/model.pt")
print("Training complete!")
""")

# Create a small data.csv locally
with open("data.csv", "w") as f:
    f.write("sepal_length,sepal_width,petal_length,petal_width,species\\n")
    f.write("5.1,3.5,1.4,0.2,0\\n")
    f.write("4.9,3.0,1.4,0.2,0\\n")
    f.write("7.0,3.2,4.7,1.4,1\\n")
    f.write("6.4,3.2,4.5,1.5,1\\n")
    f.write("6.3,3.3,6.0,2.5,2\\n")
    f.write("5.8,2.7,5.1,1.9,2\\n")

Step 2: Run on remote GPU with S3 output

result = manager.run_and_stop(
    script="python train.py",
    gpu_type="nvidia-rtx-3090",
    input_files=["train.py", "data.csv"],    # local files uploaded to remote session
    output_files=["model.pt"],                # remote files uploaded to S3 after execution
    s3_bucket="my-bucket",
    s3_prefix="training-results/run-001/",
    s3_access_key="YOUR_S3_ACCESS_KEY",
    s3_secret_key="YOUR_S3_SECRET_KEY",
    s3_endpoint="https://s3.amazonaws.com",   # optional, default: AWS S3
    s3_region="us-east-1"                     # optional, default: us-east-1
)

print(f"Success: {result.success}")
print(f"Exit code: {result.exit_code}")
print(f"Output: {result.stdout}")
print(f"Cost: {result.cost} EUR")
print(f"Duration: {result.duration_seconds}s")

11. Sessions resumed from custom environment

When you resume a session saved as a custom environment, the workspace must be re-downloaded from S3. The SDK handles this automatically:

session = manager.create_session(
    gpu_type="nvidia-rtx-3090",
    environment_id="3a07d1e9-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
    vcpu=8, ram=32
)
# Displays a progress bar:
# Workspace restore [========............] 28.4%  3.12/10.98 GB  ETA 124s

# Check progress manually
session = manager.get_session("0e4c713a")
print(f"Ready: {session.ready}")
if session.workspace_sync and session.workspace_sync.get("in_progress"):
    print(f"Progress: {session.workspace_sync['pct']:.0f}%")

12. Notifications and costs

Send an email

# Send to yourself (default: your account email)
manager.send_email(
    subject="Training Complete",
    message="Accuracy: 95%, Model saved to S3"
)

# Send to a specific recipient
manager.send_email(
    subject="Training Complete",
    message="Accuracy: 95%",
    to="colleague@company.com"
)

Generate an SDK key (sk_live_)

sdk_key = manager.generate_sdk_key("0e4c713a", name="My SDK key")
print(f"SDK key: {sdk_key}")

# Use with the clouditia SDK
from clouditia import GPUSession
session_live_gpu = GPUSession(api_key=sdk_key)
result = session_live_gpu.run("print('Hello!')")

Cost of multiple sessions

# Specific sessions
costs = manager.get_sessions_cost(["0e4c713a", "f0b09214"])
print(f"Total cost: {costs['total_cost']} EUR")

# All active sessions
active_costs = manager.get_active_sessions_cost()
print(f"Active sessions: {active_costs['session_count']}")
print(f"Total cost: {active_costs['total_cost']} EUR")

Error handling

from clouditia_manager import (
    GPUManager,
    AuthenticationError,
    SessionNotFoundError,
    InsufficientResourcesError,
    APIError
)

try:
    manager = GPUManager(api_key="sk_compute_xxxxx")
    session = manager.create_session(gpu_type="nvidia-rtx-4090")
except AuthenticationError:
    print("Invalid API key")
except InsufficientResourcesError:
    print("No GPUs available")
except SessionNotFoundError:
    print("Session not found")
except APIError as e:
    print(f"API error: {e}")

API Reference

Object attributes

GPUSession

GPUInventory

Datacenter

LambdaResult

QueueJob

License

MIT License