SDK: Python
Copy the file below into your repository (for example constellation_client.py). Its only dependency is requests. It mirrors the TypeScript client exactly:
- Auth via the
x-api-keyheader on every request. ingest_telemetrysplits any list into batches of at most 1000 records (the server cap), posts them in order, and returns one aggregated result with rejected indices mapped back into your original list.predictionsenforces the 100-idlink_idscap before the request is sent.- Non-2xx responses raise
ConstellationApiErrorwithstatus,code(theerrorfield from the response envelope:auth_lockout,rate_limited,batch_too_large,too_many_link_ids,telemetry_unavailable, and so on),retry_after_secondswhen provided, and the parsed body. - One automatic retry for 429 and 503, honoring
Retry-After, capped bymax_retry_wait_seconds(default 30) so a 900-secondauth_lockoutfails fast.
"""Vendorable single-file client for the ConstellationOS API. Requires: requests."""
import time
from dataclasses import asdict, dataclass, field
from typing import Any, Optional, Union
import requests
MAX_BATCH = 1000
MAX_LINK_IDS = 100
TOPOLOGY_MEASUREMENTS = ("link", "satellite", "ground_station", "weather", "telemetry")
HEALTH_SUBSYSTEMS = ("telemetry", "topology", "predictions")
@dataclass
class TelemetryRecord:
measurement: str
time: str # ISO 8601 timestamp, e.g. "2026-07-09T12:00:00Z"
tags: dict[str, str] = field(default_factory=dict)
fields: dict[str, Union[float, int, str]] = field(default_factory=dict)
@dataclass
class IngestResult:
batches: list[dict[str, Any]] # one TelemetryWriteAck per POST
accepted_count: int
rejected_count: int
rejected_indices: list[int] # indices into the original records list
class ConstellationApiError(Exception):
def __init__(self, status, code, message, body=None, retry_after_seconds=None):
super().__init__(message)
self.status = status
self.code = code
self.body = body
self.retry_after_seconds = retry_after_seconds
class ConstellationClient:
def __init__(
self,
base_url: str,
api_key: str,
batch_size: int = MAX_BATCH,
max_retry_wait_seconds: float = 30.0,
timeout: float = 30.0,
):
self.base_url = base_url.rstrip("/")
self.batch_size = max(1, min(batch_size, MAX_BATCH))
self.max_retry_wait_seconds = max_retry_wait_seconds
self.timeout = timeout
self._session = requests.Session()
self._session.headers["x-api-key"] = api_key
def health(self) -> Any:
"""GET /health. Open endpoint; 200 whenever the API is reachable."""
return self._request("GET", "/health")
def subsystem_health(self, name: str) -> dict[str, Any]:
"""GET /health/{name}. 503 means degraded; that is an answer, not an
exception, so this never raises on 503."""
if name not in HEALTH_SUBSYSTEMS:
raise ValueError(f"subsystem must be one of {HEALTH_SUBSYSTEMS}")
resp = self._session.get(f"{self.base_url}/health/{name}", timeout=self.timeout)
return {
"subsystem": name,
"healthy": resp.ok,
"status": resp.status_code,
"body": _parse_body(resp),
}
def topology(
self,
as_of_utc: Optional[str] = None,
measurement: Optional[str] = None,
freshness_seconds: Optional[float] = None,
) -> dict[str, Any]:
"""GET /topology: latest-sample-per-entity projection over your telemetry."""
params: list[tuple[str, str]] = []
if as_of_utc is not None:
params.append(("as_of_utc", as_of_utc))
if measurement is not None:
params.append(("measurement", measurement))
if freshness_seconds is not None:
params.append(("freshness_seconds", str(freshness_seconds)))
return self._request("GET", "/topology", params=params)
def ingest_telemetry(self, records: list) -> IngestResult:
"""POST /telemetry with automatic batching (server cap: 1000 records/request).
Accepts TelemetryRecord instances or plain dicts."""
payload = [asdict(r) if isinstance(r, TelemetryRecord) else r for r in records]
batches: list[dict[str, Any]] = []
rejected_indices: list[int] = []
for offset in range(0, len(payload), self.batch_size):
chunk = payload[offset : offset + self.batch_size]
ack = self._request("POST", "/telemetry", json_body=chunk)
batches.append(ack)
for i in ack.get("rejected_indices") or []:
rejected_indices.append(offset + i)
return IngestResult(
batches=batches,
accepted_count=sum(a["accepted_count"] for a in batches),
rejected_count=sum(a["rejected_count"] for a in batches),
rejected_indices=rejected_indices,
)
def predictions(
self,
link_ids: list[str],
model_family: str = "snr",
horizon_minutes: Optional[int] = None,
as_of_utc: Optional[str] = None,
live: bool = False,
) -> dict[str, Any]:
"""GET /predictions. Cached reads return a PredictionSet; live=True returns
an InferencePayload with full provenance. The 100-id cap is enforced here."""
if len(link_ids) > MAX_LINK_IDS:
raise ConstellationApiError(
400,
"too_many_link_ids",
f"link_ids cap is {MAX_LINK_IDS}, received {len(link_ids)}",
body={"error": "too_many_link_ids", "max": MAX_LINK_IDS, "received": len(link_ids)},
)
params: list[tuple[str, str]] = [("model_family", model_family)]
params.extend(("link_ids", link_id) for link_id in link_ids)
if horizon_minutes is not None:
params.append(("horizon_minutes", str(horizon_minutes)))
if as_of_utc is not None:
params.append(("as_of_utc", as_of_utc))
if live:
params.append(("live", "true"))
return self._request("GET", "/predictions", params=params)
def _request(self, method, path, params=None, json_body=None):
url = self.base_url + path
resp = self._session.request(method, url, params=params, json=json_body, timeout=self.timeout)
if resp.status_code in (429, 503):
wait = _retry_after(resp)
if wait <= self.max_retry_wait_seconds:
time.sleep(wait)
resp = self._session.request(method, url, params=params, json=json_body, timeout=self.timeout)
if resp.ok:
return _parse_body(resp)
raise _to_error(resp)
def _parse_body(resp):
if not resp.content:
return None
try:
return resp.json()
except ValueError:
return resp.text
def _retry_after(resp) -> float:
body = _parse_body(resp)
if isinstance(body, dict) and isinstance(body.get("retry_after_seconds"), (int, float)):
return float(body["retry_after_seconds"])
header = resp.headers.get("Retry-After")
if header is not None:
try:
return float(header)
except ValueError:
pass
return 1.0
def _to_error(resp) -> ConstellationApiError:
body = _parse_body(resp)
code = body.get("error") if isinstance(body, dict) else None
code = code if isinstance(code, str) else f"http_{resp.status_code}"
retry = body.get("retry_after_seconds") if isinstance(body, dict) else None
return ConstellationApiError(
resp.status_code,
code,
f"ConstellationOS API error {code} (HTTP {resp.status_code})",
body=body,
retry_after_seconds=retry,
)
Usage
Construct a client
import os
from constellation_client import ConstellationApiError, ConstellationClient, TelemetryRecord
client = ConstellationClient(
base_url="https://api.constellation.space",
api_key=os.environ["CONSTELLATION_API_KEY"],
)
Ingest telemetry (auto-batched)
from datetime import datetime, timezone
now = datetime.now(timezone.utc).isoformat()
result = client.ingest_telemetry([
TelemetryRecord(
measurement="link",
time=now,
tags={"entity_id": "gs-madrid--sat-041"},
fields={"snr_db": 12.4, "elevation_deg": 34.1},
),
# ...any number of records; the client splits into batches of at most 1000
])
print(f"accepted={result.accepted_count} rejected={result.rejected_count}")
if result.rejected_indices:
print("rejected input indices:", result.rejected_indices)
Read topology
topo = client.topology(measurement="link", freshness_seconds=3600)
for entity in topo["entities"]:
print(entity["entity_id"], entity["observed_at"], entity["fields"])
Predictions, cached and live
cached = client.predictions(link_ids=["gs-madrid--sat-041"], horizon_minutes=15)
live = client.predictions(link_ids=["gs-madrid--sat-041"], live=True)
if "provenance" in live:
prov = live["provenance"]
print(prov["served_from"], prov["model_version"])
for p in live["predictions"]:
print(p["link_id"], p["horizon_minutes"], p["value"]["p50"])
Handle errors
try:
client.ingest_telemetry(records)
except ConstellationApiError as err:
if err.code == "auth_lockout":
print(f"locked out; retry after {err.retry_after_seconds}s")
raise
Notes
- The request body cap is 1 MB. If 1000 of your records exceed 1 MB serialized, the server returns 413
request_too_large; lowerbatch_sizein the constructor. tenant_keyis stamped server-side from your API key; never include it in records.- Requires Python 3.9+ for the builtin generic type hints used in the dataclasses.