Add workflow management

Author: paulnoirel

libs/labelbox/src/labelbox/__init__.py

@@ -101,3 +101,4 @@
 from labelbox.schema.taskstatus import TaskStatus
 from labelbox.schema.api_key import ApiKey
 from labelbox.schema.timeunit import TimeUnit
+from labelbox.schema.workflow import ProjectWorkflow

libs/labelbox/src/labelbox/schema/project.py

@@ -59,6 +59,7 @@
     ProjectOverview,
     ProjectOverviewDetailed,
 )
+from labelbox.schema.workflow import ProjectWorkflow
 from labelbox.schema.resource_tag import ResourceTag
 from labelbox.schema.task import Task
 from labelbox.schema.task_queue import TaskQueue
@@ -1702,6 +1703,41 @@ def get_labeling_service_dashboard(self) -> LabelingServiceDashboard:
         """
         return LabelingServiceDashboard.get(self.client, self.uid)
 
+    def get_workflow(self):
+        """Get the workflow configuration for this project.
+
+        Workflows are automatically created when projects are created.
+
+        Returns:
+            ProjectWorkflow: A ProjectWorkflow object containing the project workflow information.
+        """
+        from labelbox.schema.workflow import ProjectWorkflow
+
+        return ProjectWorkflow.get_workflow(self.client, self.uid)
+
+    def clone_workflow_from(self, source_project_id: str) -> "ProjectWorkflow":
+        """Clones a workflow from another project to this project.
+
+        Args:
+            source_project_id (str): The ID of the project to clone the workflow from
+
+        Returns:
+            ProjectWorkflow: The cloned workflow in this project
+        """
+        from labelbox.schema.workflow import ProjectWorkflow
+
+        # Get the source workflow
+        source_workflow = ProjectWorkflow.get_workflow(
+            self.client, source_project_id
+        )
+
+        # Use copy_workflow_structure to clone the workflow
+        return ProjectWorkflow.copy_workflow_structure(
+            source_workflow=source_workflow,
+            target_client=self.client,
+            target_project_id=self.uid,
+        )
+
 
 class ProjectMember(DbObject):
     user = Relationship.ToOne("User", cache=True)

libs/labelbox/src/labelbox/schema/workflow/__init__.py

@@ -0,0 +1,84 @@
+"""
+This module contains classes for managing project workflows in Labelbox.
+It provides strongly-typed classes for nodes, edges, and workflow configuration.
+"""
+
+# Import all workflow classes to expose them at the package level
+from labelbox.schema.workflow.enums import (
+    WorkflowDefinitionId,
+    NodeOutput,
+    FilterField,
+    FilterOperator,
+)
+from labelbox.schema.workflow.filters import (
+    Filter,
+    UserFilter,
+    MetadataFilter,
+    TimeRangeFilter,
+    DatasetFilter,
+    AnnotationFilter,
+    NLSearchFilter,
+)
+from labelbox.schema.workflow.base import (
+    BaseWorkflowNode,
+    NodePosition,
+)
+from labelbox.schema.workflow.nodes import (
+    InitialLabelingNode,
+    InitialReworkNode,
+    ReviewNode,
+    ReworkNode,
+    LogicNode,
+    DoneNode,
+    CustomReworkNode,
+    UnknownWorkflowNode,
+)
+from labelbox.schema.workflow.edges import (
+    WorkflowEdge,
+    WorkflowEdgeFactory,
+)
+from labelbox.schema.workflow.graph import ProjectWorkflowGraph
+from labelbox.schema.workflow.project_filter import ProjectWorkflowFilter
+from labelbox.schema.workflow.workflow import ProjectWorkflow
+
+# Re-export key classes at the module level
+__all__ = [
+    "WorkflowDefinitionId",
+    "NodeOutput",
+    "FilterField",
+    "FilterOperator",
+    "BaseWorkflowNode",
+    "NodePosition",
+    "Filter",
+    "UserFilter",
+    "MetadataFilter",
+    "TimeRangeFilter",
+    "DatasetFilter",
+    "AnnotationFilter",
+    "NLSearchFilter",
+    "InitialLabelingNode",
+    "InitialReworkNode",
+    "ReviewNode",
+    "ReworkNode",
+    "LogicNode",
+    "DoneNode",
+    "CustomReworkNode",
+    "UnknownWorkflowNode",
+    "WorkflowEdge",
+    "WorkflowEdgeFactory",
+    "ProjectWorkflow",
+    "ProjectWorkflowGraph",
+    "ProjectWorkflowFilter",
+]
+
+# Define a mapping of node types for backward compatibility
+NODE_TYPE_MAP = {
+    WorkflowDefinitionId.InitialLabelingTask: InitialLabelingNode,
+    WorkflowDefinitionId.InitialReworkTask: InitialReworkNode,
+    WorkflowDefinitionId.ReviewTask: ReviewNode,
+    WorkflowDefinitionId.SendToRework: ReworkNode,
+    WorkflowDefinitionId.Logic: LogicNode,
+    WorkflowDefinitionId.Done: DoneNode,
+    WorkflowDefinitionId.CustomReworkTask: CustomReworkNode,
+    WorkflowDefinitionId.Unknown: UnknownWorkflowNode,
+}

libs/labelbox/src/labelbox/schema/workflow/base.py

@@ -0,0 +1,95 @@
+"""Base classes for Project Workflow nodes and common utilities."""
+
+import logging
+from typing import Dict, List, Any, Optional
+from pydantic import BaseModel, ConfigDict, Field
+
+from labelbox.schema.workflow.enums import WorkflowDefinitionId, NodeOutput
+
+logger = logging.getLogger(__name__)
+
+
+class NodePosition(BaseModel):
+    """Position of a node in the workflow canvas."""
+
+    x: float
+    y: float
+
+
+class BaseWorkflowNode(BaseModel):
+    """Base class for all workflow node types."""
+
+    id: str
+    position: NodePosition
+    raw_data: Dict[str, Any] = Field(
+        default_factory=dict,
+        description="Store the original dict for reference",
+    )
+    inputs: List[str] = Field(
+        default_factory=list, description="Input connection identifiers"
+    )
+    output_if: Optional[str] = Field(
+        default=None, description="Primary/if output connection identifier"
+    )
+    output_else: Optional[str] = Field(
+        default=None, description="Secondary/else output connection identifier"
+    )
+    definition_id: WorkflowDefinitionId = Field(
+        default=WorkflowDefinitionId.Unknown,
+        frozen=True,
+        alias="definitionId",
+    )
+    model_config = ConfigDict(
+        arbitrary_types_allowed=True,
+        populate_by_name=True,
+    )
+
+    def _update_node_in_workflow(self) -> None:
+        """Update the node data in the workflow's config."""
+        workflow = self.raw_data.get("_workflow")
+        if workflow and hasattr(workflow, "config"):
+            for node_data in workflow.config.get("nodes", []):
+                if node_data.get("id") == self.id:
+                    # Update the workflow config with current node state
+                    if hasattr(self, "label"):
+                        node_data["label"] = self.label
+                    if hasattr(self, "custom_fields"):
+                        node_data["customFields"] = self.custom_fields
+                    if (
+                        hasattr(self, "instructions")
+                        and self.instructions is not None
+                    ):
+                        # Ensure customFields exists
+                        if "customFields" not in node_data:
+                            node_data["customFields"] = {}
+                        node_data["customFields"]["description"] = (
+                            self.instructions
+                        )
+                    break
+
+    @property
+    def name(self) -> Optional[str]:
+        """Get the node's name (label)."""
+        return self.raw_data.get("label")
+
+    @name.setter
+    def name(self, value: str) -> None:
+        """Set the node's name (label)."""
+        if hasattr(self, "label"):
+            self.label = value
+            self.raw_data["label"] = value
+            self._update_node_in_workflow()
+
+    @property
+    def supported_outputs(self) -> List[NodeOutput]:
+        """Returns the list of supported output types for this node."""
+        return []
+
+    @property
+    def config(self) -> Dict[str, Any]:
+        """Returns the node's configuration."""
+        return self.raw_data
+
+    def __repr__(self):
+        """String representation of the node."""
+        return f"{self.__class__.__name__}(id={self.id}, name={self.name})"