from __future__ import annotations
"""Deployments resource for the CorePlexML SDK."""
from typing import Union
from coreplexml._http import HTTPClient
[docs]
class DeploymentsResource:
"""Deploy models to production endpoints.
Deployments create REST API endpoints for real-time predictions,
with support for staging/production stages and canary rollouts.
"""
def __init__(self, http: HTTPClient):
self._http = http
@staticmethod
def _normalize_deployment(payload: dict) -> dict:
"""Normalize API payloads to a direct deployment object."""
if not isinstance(payload, dict):
return {}
dep = payload.get("deployment")
if isinstance(dep, dict):
out = dict(dep)
# Preserve model payload when present for callers that need it.
if isinstance(payload.get("model"), dict):
out["_model"] = payload["model"]
return out
return payload
[docs]
def list(self, project_id: str, limit: int = 50, offset: int = 0) -> dict:
"""List deployments for a project.
Args:
project_id: UUID of the project.
limit: Maximum results (default 50).
offset: Pagination offset.
Returns:
Dictionary with ``items`` list and ``total`` count.
"""
return self._http.get(f"/api/mlops/projects/{project_id}/deployments", params={"limit": limit, "offset": offset})
[docs]
def create(self, project_id: str, model_id: str, name: str, stage: str = "staging", config: dict | None = None) -> dict:
"""Create a new deployment.
Args:
project_id: UUID of the project.
model_id: UUID of the model to deploy.
name: Deployment name.
stage: Deployment stage -- ``staging`` or ``production`` (default ``staging``).
config: Optional deployment configuration.
Returns:
Created deployment dictionary.
"""
body: dict = {"model_id": model_id, "name": name, "stage": stage}
if config:
body["config"] = config
data = self._http.post(f"/api/mlops/projects/{project_id}/deployments", json=body)
return self._normalize_deployment(data)
[docs]
def get(self, deployment_id: str) -> dict:
"""Get deployment details.
Args:
deployment_id: UUID of the deployment.
Returns:
Deployment dictionary.
"""
data = self._http.get(f"/api/mlops/deployments/{deployment_id}")
return self._normalize_deployment(data)
[docs]
def predict(self, deployment_id: str, inputs: Union[dict, list], options: dict | None = None) -> dict:
"""Make predictions via a deployed model endpoint.
Args:
deployment_id: UUID of the deployment.
inputs: Feature values -- a dict or list of dicts.
options: Optional prediction options.
Returns:
Prediction results dictionary.
"""
body = {"inputs": inputs, "options": options or {}}
data = self._http.post(f"/api/mlops/deployments/{deployment_id}/predict", json=body)
# Convenience aliases for single-row predictions used in quickstart docs.
if isinstance(inputs, dict) and isinstance(data, dict):
preds = data.get("predictions")
if isinstance(preds, list) and preds:
first = preds[0] if isinstance(preds[0], dict) else {}
if "prediction" in first and "prediction" not in data:
data["prediction"] = first.get("prediction")
if "probability" in first and "probability" not in data:
data["probability"] = first.get("probability")
if "probabilities" in first and "probabilities" not in data:
data["probabilities"] = first.get("probabilities")
return data
[docs]
def rollback(
self,
deployment_id: str,
to_deployment_id: str | None = None,
to_model_id: str | None = None,
) -> dict:
"""Rollback a deployment to the previous version.
Args:
deployment_id: UUID of the deployment.
to_deployment_id: Optional target deployment UUID.
to_model_id: Optional target model UUID.
Returns:
Updated deployment dictionary.
"""
body: dict = {}
if to_deployment_id:
body["to_deployment_id"] = to_deployment_id
if to_model_id:
body["to_model_id"] = to_model_id
data = self._http.post(f"/api/mlops/deployments/{deployment_id}/rollback", json=body)
return self._normalize_deployment(data)
[docs]
def deactivate(self, deployment_id: str) -> dict:
"""Deactivate a deployment.
Args:
deployment_id: UUID of the deployment.
Returns:
Updated deployment dictionary.
"""
data = self._http.post(f"/api/mlops/deployments/{deployment_id}/deactivate")
return self._normalize_deployment(data)
[docs]
def drift(self, deployment_id: str) -> dict:
"""Get drift detection results for a deployment.
Args:
deployment_id: UUID of the deployment.
Returns:
Drift metrics dictionary.
"""
return self._http.get(f"/api/mlops/deployments/{deployment_id}/drift")