4.1 updates and pipelines route

Author: jake-valsamis

src/codeocean/capsule.py

@@ -5,7 +5,13 @@
 from typing import Optional, Iterator
 from requests_toolbelt.sessions import BaseUrlSession
 
-from codeocean.components import Ownership, SortOrder, SearchFilter, Permissions
+from codeocean.components import (
+    Ownership,
+    SortOrder,
+    SearchFilter,
+    Permissions,
+    AppPanel,
+)
 from codeocean.computation import Computation
 from codeocean.data_asset import DataAssetAttachParams, DataAssetAttachResults
 from codeocean.enum import StrEnum
@@ -26,29 +32,6 @@ 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:
@@ -272,199 +255,6 @@ 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(
-        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."""
@@ -547,24 +337,3 @@ def search_capsules_iterator(self, search_params: CapsuleSearchParams) -> Iterat
                 return
 
             params["next_token"] = response.next_token
-
-    def search_pipelines(self, search_params: CapsuleSearchParams) -> CapsuleSearchResults:
-        """Search for pipelines with filtering, sorting, and pagination
-        options."""
-        res = self.client.post("pipelines/search", json=search_params.to_dict())
-
-        return CapsuleSearchResults.from_dict(res.json())
-
-    def search_pipelines_iterator(self, search_params: CapsuleSearchParams) -> Iterator[Capsule]:
-        """Iterate through all pipelines matching search criteria with automatic pagination."""
-        params = search_params.to_dict()
-        while True:
-            response = self.search_pipelines(search_params=CapsuleSearchParams(**params))
-
-            for result in response.results:
-                yield result
-
-            if not response.has_more:
-                return
-
-            params["next_token"] = response.next_token

src/codeocean/client.py

@@ -12,6 +12,7 @@
 from codeocean.custom_metadata import CustomMetadataSchema
 from codeocean.data_asset import DataAssets
 from codeocean.error import Error
+from codeocean.pipeline import Pipelines
 
 
 @dataclass
@@ -55,6 +56,7 @@ def __post_init__(self):
         self.computations = Computations(client=self.session)
         self.custom_metadata = CustomMetadataSchema(client=self.session)
         self.data_assets = DataAssets(client=self.session)
+        self.pipelines = Pipelines(client=self.session)
 
     def _error_handler(self, response, *args, **kwargs):
         try: