from __future__ import annotations
"""Projects resource for the CorePlexML SDK."""
from coreplexml._http import HTTPClient
[docs]
class ProjectsResource:
"""Manage CorePlexML projects.
Projects are the top-level organizational unit. Each project
contains datasets, experiments, models, and deployments.
"""
def __init__(self, http: HTTPClient):
self._http = http
@staticmethod
def _normalize_project(payload: dict) -> dict:
"""Normalize API payloads to a direct project object."""
if not isinstance(payload, dict):
return {}
project = payload.get("project")
if isinstance(project, dict):
out = dict(project)
if "id" not in out and payload.get("project_id"):
out["id"] = payload["project_id"]
return out
return payload
[docs]
def list(self, limit: int = 50, offset: int = 0, search: str | None = None) -> dict:
"""List all projects accessible to the authenticated user.
Args:
limit: Maximum number of projects to return (default 50).
offset: Number of projects to skip for pagination.
search: Optional search query to filter by name.
Returns:
Dictionary with ``items`` list and ``total`` count.
Raises:
AuthenticationError: If the API key is invalid.
"""
params = {"limit": limit, "offset": offset}
if search:
params["search"] = search
return self._http.get("/api/projects", params=params)
[docs]
def create(self, name: str, description: str = "") -> dict:
"""Create a new project.
Args:
name: Project name (must be non-empty).
description: Optional project description.
Returns:
Created project dictionary with ``id``, ``name``, etc.
Raises:
ValidationError: If the name is empty.
"""
data = self._http.post("/api/projects", json={"name": name, "description": description})
return self._normalize_project(data)
[docs]
def get(self, project_id: str) -> dict:
"""Get project details by ID.
Args:
project_id: UUID of the project.
Returns:
Project dictionary.
Raises:
NotFoundError: If the project does not exist.
"""
data = self._http.get(f"/api/projects/{project_id}")
return self._normalize_project(data)
[docs]
def update(self, project_id: str, name: str | None = None, description: str | None = None) -> dict:
"""Update project name and/or description.
If ``name`` is not provided, the current name is preserved automatically
(the API requires ``name`` on every update).
Args:
project_id: UUID of the project.
name: New project name (optional — keeps current if omitted).
description: New project description (optional).
Returns:
Updated project dictionary.
"""
# API requires name on every PUT — fetch current if not provided
if name is None:
current = self._http.get(f"/api/projects/{project_id}")
project = self._normalize_project(current)
name = project.get("name", "")
body: dict = {"name": name}
if description is not None:
body["description"] = description
data = self._http.put(f"/api/projects/{project_id}", json=body)
return self._normalize_project(data)
[docs]
def delete(self, project_id: str) -> dict:
"""Delete a project and all associated resources.
Args:
project_id: UUID of the project.
Returns:
Empty dictionary on success.
"""
return self._http.delete(f"/api/projects/{project_id}")
[docs]
def members(self, project_id: str) -> dict:
"""List project members.
Args:
project_id: UUID of the project.
Returns:
Dictionary with paginated ``items`` list plus ``total``, ``limit``, and ``offset``.
"""
return self._http.get(f"/api/projects/{project_id}/members")
[docs]
def add_member(self, project_id: str, email: str, role: str = "editor") -> dict:
"""Add a member to a project.
Args:
project_id: UUID of the project.
email: Email address of the user to add.
role: Member role -- ``viewer``, ``editor``, or ``admin`` (default ``editor``).
Returns:
Created membership dictionary.
"""
return self._http.post(f"/api/projects/{project_id}/members", json={"email": email, "role": role})
[docs]
def remove_member(self, project_id: str, member_id: str) -> dict:
"""Remove a member from a project.
Args:
project_id: UUID of the project.
member_id: UUID of the membership record.
Returns:
Empty dictionary on success.
"""
return self._http.delete(f"/api/projects/{project_id}/members/{member_id}")
[docs]
def timeline(self, project_id: str) -> dict:
"""Get project activity timeline.
Args:
project_id: UUID of the project.
Returns:
Dictionary with ``events`` list.
"""
return self._http.get(f"/api/projects/{project_id}/timeline")