"""Deployments resource for the CorePlexML SDK."""
from __future__ import annotations
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,
*,
traffic_percent: int = 100,
privacy_policy_id: str | None = None,
privacy_anonymize_logs: bool | None = None,
privacy_anonymize_response: bool | None = None,
privacy_column_map: dict[str, str] | None = None,
privacy_threshold: float | None = None,
decision_threshold: float | None = None,
positive_class: str = "1",
quality_gate_override: bool = False,
quality_gate_override_reason: str | 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: Backward-compatible mapping of supported deployment fields.
Unknown fields are rejected instead of being silently ignored.
traffic_percent: Percentage of traffic routed to the deployment.
privacy_policy_id: Optional runtime privacy policy UUID.
privacy_anonymize_logs: Anonymize stored inference logs when enabled.
privacy_anonymize_response: Anonymize prediction responses when enabled.
privacy_column_map: Optional input-to-policy column mapping.
privacy_threshold: Privacy detector confidence threshold (0 to 1).
decision_threshold: Positive-class operating threshold captured by
the deployment. Defaults to the model operating point.
positive_class: Class label treated as the positive outcome.
quality_gate_override: Explicitly override a blocked quality gate.
quality_gate_override_reason: Accountable business/technical reason for
the override. Required by the server when an override is used.
Returns:
Created deployment dictionary.
"""
supported_config_fields = {
"traffic_percent",
"privacy_policy_id",
"privacy_anonymize_logs",
"privacy_anonymize_response",
"privacy_column_map",
"privacy_threshold",
"decision_threshold",
"positive_class",
"quality_gate_override",
"quality_gate_override_reason",
}
if config:
unknown = set(config) - supported_config_fields
if unknown:
raise ValueError(
"Unsupported deployment config fields: "
+ ", ".join(sorted(unknown))
)
traffic_percent = config.get("traffic_percent", traffic_percent)
privacy_policy_id = config.get("privacy_policy_id", privacy_policy_id)
privacy_anonymize_logs = config.get(
"privacy_anonymize_logs", privacy_anonymize_logs
)
privacy_anonymize_response = config.get(
"privacy_anonymize_response", privacy_anonymize_response
)
privacy_column_map = config.get("privacy_column_map", privacy_column_map)
privacy_threshold = config.get("privacy_threshold", privacy_threshold)
decision_threshold = config.get("decision_threshold", decision_threshold)
positive_class = config.get("positive_class", positive_class)
quality_gate_override = config.get(
"quality_gate_override", quality_gate_override
)
quality_gate_override_reason = config.get(
"quality_gate_override_reason", quality_gate_override_reason
)
body: dict = {
"model_id": model_id,
"name": name,
"stage": stage,
"traffic_percent": traffic_percent,
"quality_gate_override": quality_gate_override,
"positive_class": positive_class,
}
optional_fields = {
"privacy_policy_id": privacy_policy_id,
"privacy_anonymize_logs": privacy_anonymize_logs,
"privacy_anonymize_response": privacy_anonymize_response,
"privacy_column_map": privacy_column_map,
"privacy_threshold": privacy_threshold,
"decision_threshold": decision_threshold,
"quality_gate_override_reason": quality_gate_override_reason,
}
body.update(
{key: value for key, value in optional_fields.items() if value is not None}
)
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: 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")
[docs]
def run_drift(self, deployment_id: str) -> dict:
"""Enqueue a drift analysis for a deployment."""
return self._http.post(f"/api/mlops/deployments/{deployment_id}/drift/run")
[docs]
def inference_logs(self, deployment_id: str, limit: int = 100) -> dict:
return self._http.get(
f"/api/mlops/deployments/{deployment_id}/inference-logs",
params={"limit": limit},
)
[docs]
def api_docs(self, deployment_id: str) -> dict:
return self._http.get(f"/api/mlops/deployments/{deployment_id}/api-docs")
[docs]
def rollback_history(self, deployment_id: str) -> dict:
return self._http.get(
f"/api/mlops/deployments/{deployment_id}/rollback-history"
)