feat: moves task effects (#208)

canvas_sdk/effects/task/task.py

@@ -0,0 +1,99 @@
+from datetime import datetime
+from enum import Enum
+from typing import Any, cast
+
+from canvas_sdk.effects.base import EffectType, _BaseEffect
+
+
+class TaskStatus(Enum):
+    """TaskStatus."""
+
+    COMPLETED = "COMPLETED"
+    CLOSED = "CLOSED"
+    OPEN = "OPEN"
+
+
+class AddTask(_BaseEffect):
+    """
+    An Effect that will create a Task in Canvas.
+    """
+
+    class Meta:
+        effect_type = EffectType.CREATE_TASK
+        apply_required_fields = ("title",)
+
+    assignee_id: str | None = None
+    patient_id: str | None = None
+    title: str | None = None
+    due: datetime | None = None
+    status: TaskStatus = TaskStatus.OPEN
+    labels: list[str] = []
+
+    @property
+    def values(self) -> dict[str, Any]:
+        """The values for Task addition."""
+        return {
+            "patient": {"id": self.patient_id},
+            "due": self.due.isoformat() if self.due else None,
+            "assignee": {"id": self.assignee_id},
+            "title": self.title,
+            "status": self.status.value,
+            "labels": self.labels,
+        }
+
+
+class AddTaskComment(_BaseEffect):
+    """
+    An Effect that will create a Task Comment on a Task.
+    """
+
+    class Meta:
+        effect_type = EffectType.CREATE_TASK_COMMENT
+        apply_required_fields = (
+            "body",
+            "task_id",
+        )
+
+    body: str | None = None
+    task_id: str | None = None
+
+    @property
+    def values(self) -> dict[str, Any]:
+        """The values for adding a task comment."""
+        return {"task": {"id": self.task_id}, "body": self.body}
+
+
+class UpdateTask(_BaseEffect):
+    """
+    An Effect that will update a Task in Canvas.
+    """
+
+    class Meta:
+        effect_type = EffectType.UPDATE_TASK
+        apply_required_fields = ("id",)
+
+    id: str | None = None
+    assignee_id: str | None = None
+    patient_id: str | None = None
+    title: str | None = None
+    due: datetime | None = None
+    status: TaskStatus = TaskStatus.OPEN
+    labels: list[str] = []
+
+    @property
+    def values(self) -> dict[str, Any]:
+        """The values for adding a task comment."""
+        value_dict: dict[str, Any] = {}
+        # Only add the fields that have been explicitly set on the model (exclude_unset=True).
+        # Oherwise, the effect interpreter will set values to null based on their defaults.
+        set_fields = self.model_dump(exclude_unset=True)
+        for field, val in set_fields.items():
+            if field.endswith("_id"):
+                value_dict[field.split("_")[0]] = {"id": val}
+            elif field == "due" and val is not None:
+                value_dict[field] = cast(datetime, val).isoformat()
+            elif field == "status":
+                value_dict[field] = cast(TaskStatus, val).value
+            else:
+                value_dict[field] = getattr(self, field)
+        return value_dict

canvas_sdk/protocols/base.py

@@ -1,8 +1,5 @@
 from typing import Any
 
-import arrow
-
-from canvas_sdk.data.client import GQL_CLIENT
 from canvas_sdk.handlers.base import BaseHandler
 
 
@@ -11,13 +8,4 @@ class BaseProtocol(BaseHandler):
     The class that protocols inherit from.
     """
 
-    def _beta_run_gql_query(self, query: str, variables: dict | None = None) -> dict[str, Any]:
-        return GQL_CLIENT.query(
-            query,
-            variables=variables,
-            extra_args={
-                "headers": {
-                    "Authorization": f'Bearer {self.secrets["graphql_jwt"]}',
-                },
-            },
-        )
+    pass