Spaces:
Configuration error
Configuration error
File size: 1,852 Bytes
06fed5f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 | """DataRobot client stub (control plane).
Workload/agentic operations are guarded behind WORKLOAD_API_ENABLED because
those are premium, entitlement-gated features. Set the flag ONLY if your tenant
has the corresponding entitlement.
"""
import os
import requests
BASE = os.getenv("DATAROBOT_ENDPOINT", "https://app.datarobot.com/api/v2")
TOKEN = os.getenv("DATAROBOT_API_TOKEN")
WORKLOADS_ENABLED = os.getenv("WORKLOAD_API_ENABLED", "false").lower() == "true"
def _headers():
if not TOKEN:
raise RuntimeError("DATAROBOT_API_TOKEN not set")
return {"Authorization": f"Bearer {TOKEN}"}
# --- Projects / predictive AI ---
def list_projects():
return requests.get(f"{BASE}/projects/", headers=_headers(), timeout=30).json()
def get_project(project_id: str):
return requests.get(f"{BASE}/projects/{project_id}/", headers=_headers(), timeout=30).json()
# --- Deployments (MLOps prediction) ---
def list_deployments():
return requests.get(f"{BASE}/deployments/", headers=_headers(), timeout=30).json()
def predict(deployment_id: str, data):
# Batch/prediction endpoints; see registry for exact paths
return requests.post(
f"{BASE}/predApi/deployments/{deployment_id}/predictions",
headers=_headers(), json=data, timeout=60,
).json()
# --- Workloads (premium, guarded) ---
def list_workloads():
if not WORKLOADS_ENABLED:
raise RuntimeError(
"Workload API is premium/entitlement-gated; set WORKLOAD_API_ENABLED=true only if entitled."
)
return requests.get(f"{BASE}/workloads", headers=_headers(), timeout=30).json()
def create_workload(payload: dict):
if not WORKLOADS_ENABLED:
raise RuntimeError("Workload API not enabled for this tenant")
return requests.post(f"{BASE}/workloads", headers=_headers(), json=payload, timeout=30).json()
|