Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
237 changes: 237 additions & 0 deletions src/codeocean/capsule.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,29 @@ class CapsuleSortBy(StrEnum):
Name = "name"


class AppPanelDataAssetKind(StrEnum):
"""The kind of data asset displayed in an app panel.

- 'Internal' → Data stored inside Code Ocean.
- 'External' → Data stored external to Code Ocean.
- 'Combined' → Data containing multiple external data assets.

In pipelines, a data asset can only be replaced with one of the same kind.
"""

Internal = "internal"
External = "external"
Combined = "combined"


class AppPanelParameterType(StrEnum):
"""The type of parameter displayed in an app panel."""

Text = "text"
List = "list"
File = "file"


@dataclass_json
@dataclass(frozen=True)
class OriginalCapsuleInfo:
Expand Down Expand Up @@ -81,6 +104,10 @@ class Capsule:
slug: str = dataclass_field(
metadata={"description": "Alternate capsule ID (URL-friendly identifier)"},
)
last_accessed: Optional[int] = dataclass_field(
default=None,
metadata={"description": "Capsule last accessed time (int64 timestamp)"},
)
article: Optional[dict] = dataclass_field(
default=None,
metadata={
Expand Down Expand Up @@ -214,6 +241,199 @@ class CapsuleSearchResults:
)


@dataclass_json
@dataclass(frozen=True)
class AppPanelCategories:
"""Categories for a capsule's App Panel parameters."""

id: str = dataclass_field(
metadata={"description": "Unique identifier for the category."},
)
name: str = dataclass_field(
metadata={"description": "Human-readable name of the category."},
)
description: Optional[str] = dataclass_field(
default=None,
metadata={"description": "Optional detailed description of the category."},
)
help_text: Optional[str] = dataclass_field(
default=None,
metadata={"description": "Optional help text providing guidance or additional information about the category."},
)


@dataclass_json
@dataclass(frozen=True)
class AppPanelParameters:
"""Parameters for a capsule's App Panel."""

name: str = dataclass_field(
metadata={"description": "Parameter label/display name."}
)
type: AppPanelParameterType = dataclass_field(
metadata={"description": "Type of the parameter (text, list, file)."}
)
category: Optional[str] = dataclass_field(
default=None,
metadata={"description": "ID of category the parameter belongs to."}
)
param_name: Optional[str] = dataclass_field(
default=None,
metadata={"description": "The parameter name/argument key"}
)
description: Optional[str] = dataclass_field(
default=None,
metadata={"description": "Description of the parameter."}
)
help_text: Optional[str] = dataclass_field(
default=None,
metadata={"description": "Help text for the parameter."}
)
value_type: Optional[str] = dataclass_field(
default=None,
metadata={"description": "Value type of the parameter."}
)
default_value: Optional[str] = dataclass_field(
default=None,
metadata={"description": "Default value of the parameter."}
)
required: Optional[bool] = dataclass_field(
default=None,
metadata={"description": "Indicates if the parameter is required."}
)
hidden: Optional[bool] = dataclass_field(
default=None,
metadata={"description": "Indicates if the parameter is hidden."}
)
minimum: Optional[float] = dataclass_field(
default=None,
metadata={"description": "Minimum value for the parameter."}
)
maximum: Optional[float] = dataclass_field(
default=None,
metadata={"description": "Maximum value for the parameter."}
)
pattern: Optional[str] = dataclass_field(
default=None,
metadata={"description": "Regular expression pattern for the parameter."}
)
value_options: Optional[list[str]] = dataclass_field(
default=None,
metadata={"description": "Allowed values for the parameter."}
)


@dataclass_json
@dataclass(frozen=True)
class AppPanelGeneral:
"""General information about a capsule's App Panel."""

title: Optional[str] = dataclass_field(
default=None,
metadata={"description": "Title of the App Panel."}
)
instructions: Optional[str] = dataclass_field(
default=None,
metadata={"description": "Instructions for using the App Panel."}
)
help_text: Optional[str] = dataclass_field(
default=None,
metadata={"description": "Help text for the App Panel."}
)


@dataclass_json
@dataclass(frozen=True)
class AppPanelDataAsset:
"""Data asset parameter for the App Panel."""

id: str = dataclass_field(
metadata={"description": "Unique identifier for the data asset."}
)
mount: str = dataclass_field(
metadata={"description": "Mount path of the data asset within the capsule. "
"Use this mount path to replace the currently attached data asset with your own"}
)
name: str = dataclass_field(
metadata={"description": "Display name of the data asset."}
)
kind: AppPanelDataAssetKind = dataclass_field(
metadata={"description": "Kind of the data asset (internal, external, combined)."}
)
accessible: bool = dataclass_field(
metadata={"description": "Indicates if the data asset is accessible to the user."}
)
description: Optional[str] = dataclass_field(
default=None,
metadata={"description": "Optional description of the data asset parameter."}
)
help_text: Optional[str] = dataclass_field(
default=None,
metadata={"description": "Optional help text for the data asset parameter."}
)


@dataclass_json
@dataclass(frozen=True)
class AppPanelResult:
"""Selected result files to display once the computation is complete."""

file_name: str = dataclass_field(
metadata={"description": "Name of the result file."}
)


@dataclass_json
@dataclass(frozen=True)
class AppPanelProcess:
"""Pipeline process name and its corresponding app panel (for pipelines of capsules only)"""

name: str = dataclass_field(
metadata={"description": "Name of the pipeline process."}
)
categories: Optional[AppPanelCategories] = dataclass_field(
default=None,
metadata={"description": "Categories for the pipeline process's app panel parameters."}
)
parameters: Optional[AppPanelParameters] = dataclass_field(
Comment on lines +394 to +398
Copy link

Copilot AI Aug 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The categories and parameters fields should be lists, not single objects. Based on the AppPanel class structure, these should be Optional[list[AppPanelCategories]] and Optional[list[AppPanelParameters]] respectively to match the pattern used elsewhere.

Suggested change
categories: Optional[AppPanelCategories] = dataclass_field(
default=None,
metadata={"description": "Categories for the pipeline process's app panel parameters."}
)
parameters: Optional[AppPanelParameters] = dataclass_field(
categories: Optional[list[AppPanelCategories]] = dataclass_field(
default=None,
metadata={"description": "Categories for the pipeline process's app panel parameters."}
)
parameters: Optional[list[AppPanelParameters]] = dataclass_field(

Copilot uses AI. Check for mistakes.
default=None,
metadata={"description": "Parameters for the pipeline process's app panel."}
)


@dataclass_json
@dataclass(frozen=True)
class AppPanel:
"""App Panel configuration for a capsule or pipeline, including general info, data assets,
categories, parameters, and results.
"""

general: Optional[AppPanelGeneral] = dataclass_field(
default=None,
metadata={"description": "General information about the App Panel."}
)
data_assets: Optional[list[AppPanelDataAsset]] = dataclass_field(
default=None,
metadata={"description": "List of data assets used in the App Panel."}
)
categories: Optional[list[AppPanelCategories]] = dataclass_field(
default=None,
metadata={"description": "Categories for organizing App Panel parameters."}
)
parameters: Optional[list[AppPanelParameters]] = dataclass_field(
default=None,
metadata={"description": "Parameters for the App Panel."}
)
results: Optional[list[AppPanelResult]] = dataclass_field(
default=None,
metadata={"description": "Result files to display after computation."}
)
processes: Optional[list[AppPanelProcess]] = dataclass_field(
default=None,
metadata={"description": "Pipeline processes and their App Panels."}
)


@dataclass
class Capsules:
"""Client for interacting with Code Ocean capsule APIs."""
Expand All @@ -226,6 +446,16 @@ def get_capsule(self, capsule_id: str) -> Capsule:

return Capsule.from_dict(res.json())

def delete_capsule(self, capsule_id: str):
"""Delete a capsule permanently."""
self.client.delete(f"capsules/{capsule_id}")
Comment on lines +449 to +451
Copy link

Copilot AI Aug 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The delete_capsule method should validate the HTTP response status. Without checking the response, the method will silently fail if the deletion is unsuccessful, making it difficult to detect errors.

Suggested change
def delete_capsule(self, capsule_id: str):
"""Delete a capsule permanently."""
self.client.delete(f"capsules/{capsule_id}")
res = self.client.delete(f"capsules/{capsule_id}")
res.raise_for_status()

Copilot uses AI. Check for mistakes.

def get_capsule_app_panel(self, capsule_id: str, version: Optional[int] = None) -> AppPanel:
"""Retrieve app panel information for a specific capsule by its ID."""
res = self.client.get(f"capsules/{capsule_id}/app_panel", params={"version": version} if version else None)

return AppPanel.from_dict(res.json())

def list_computations(self, capsule_id: str) -> list[Computation]:
"""Get all computations associated with a specific capsule."""
res = self.client.get(f"capsules/{capsule_id}/computations")
Expand Down Expand Up @@ -259,6 +489,13 @@ def detach_data_assets(self, capsule_id: str, data_assets: list[str]):
json=data_assets,
)

def archive_capsule(self, capsule_id: str, archive: bool):
"""Archive or unarchive a capsule to control its visibility and accessibility."""
self.client.patch(
f"capsules/{capsule_id}/archive",
params={"archive": archive},
)

def search_capsules(self, search_params: CapsuleSearchParams) -> CapsuleSearchResults:
"""Search for capsules with filtering, sorting, and pagination
options."""
Expand Down
4 changes: 3 additions & 1 deletion src/codeocean/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@

from codeocean.capsule import Capsules
from codeocean.computation import Computations
from codeocean.custom_metadata import CustomMetadataSchema
from codeocean.data_asset import DataAssets
from codeocean.error import Error

Expand Down Expand Up @@ -36,7 +37,7 @@ class CodeOcean:
agent_id: Optional[str] = None

# Minimum server version required by this SDK
MIN_SERVER_VERSION = "3.6.0"
MIN_SERVER_VERSION = "3.9.0"

def __post_init__(self):
self.session = BaseUrlSession(base_url=f"{self.domain}/api/v1/")
Expand All @@ -52,6 +53,7 @@ def __post_init__(self):

self.capsules = Capsules(client=self.session)
self.computations = Computations(client=self.session)
self.custom_metadata = CustomMetadataSchema(client=self.session)
self.data_assets = DataAssets(client=self.session)

def _error_handler(self, response, *args, **kwargs):
Expand Down
96 changes: 96 additions & 0 deletions src/codeocean/custom_metadata.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,96 @@
from __future__ import annotations

from dataclasses import dataclass, field as dataclass_field
from dataclasses_json import dataclass_json
from typing import Optional, Union
from requests_toolbelt.sessions import BaseUrlSession

from codeocean.enum import StrEnum


class CustomMetadataFieldType(StrEnum):
""" Type of the custom metadata field value. """

String = "string"
Number = "number"
Date = "date"


@dataclass_json
@dataclass(frozen=True)
class CustomMetadataFieldRange:
""" Range of valid values for a custom metadata field. """

min: Optional[float] = dataclass_field(
default=None,
metadata={"description": "Minimum valid value"}
)
max: Optional[float] = dataclass_field(
default=None,
metadata={"description": "Maximum valid value"}
)


@dataclass_json
@dataclass(frozen=True)
class CustomMetadataField:
""" Represents a custom metadata field in the Code Ocean platform. """

name: str = dataclass_field(
metadata={"description": "Name of the custom metadata field"}
)
type: CustomMetadataFieldType = dataclass_field(
metadata={"description": "Type of the custom metadata field value (string, number, date)"}
)
range: Optional[CustomMetadataFieldRange] = dataclass_field(
default=None,
metadata={"description": "Range of valid values for the field"}
)
allowed_values: Optional[Union[list[str], list[float]]] = dataclass_field(
default=None,
metadata={"description": "Allowed values for the field (item type according to field type)"}
)
multiple: Optional[bool] = dataclass_field(
default=None,
metadata={"description": "Whether multiple values are allowed"}
)
units: Optional[str] = dataclass_field(
default=None,
metadata={"description": "Units of the field value"}
)
category: Optional[str] = dataclass_field(
default=None,
metadata={"description": "Category of the field"}
)
required: Optional[bool] = dataclass_field(
default=None,
metadata={"description": "Whether the field is required"}
)


@dataclass_json
@dataclass(frozen=True)
class CustomMetadata:
""" Represents the custom metadata schema in the Code Ocean platform. """

fields: Optional[list[CustomMetadataField]] = dataclass_field(
default=None,
metadata={"description": "List of custom metadata fields"}
)
categories: Optional[list[str]] = dataclass_field(
default=None,
metadata={"description": "List of categories for custom metadata fields"}
)


@dataclass
class CustomMetadataSchema:
"""Client for getting the Code Ocean custom metadata schema."""

client: BaseUrlSession

def get_custom_metadata(self) -> CustomMetadata:
"""Retrieve the Code Ocean deployment's custom metadata schema."""
res = self.client.get("custom_metadata")

Comment on lines +92 to +95
Copy link

Copilot AI Aug 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The get_custom_metadata method should validate the HTTP response status before attempting to parse JSON. Without checking res.status_code or using res.raise_for_status(), the method may fail silently or raise unclear exceptions on API errors.

Suggested change
def get_custom_metadata(self) -> CustomMetadata:
"""Retrieve the Code Ocean deployment's custom metadata schema."""
res = self.client.get("custom_metadata")
res.raise_for_status()

Copilot uses AI. Check for mistakes.
return CustomMetadata.from_dict(res.json())