[SILO-1143] feat: add epics write endpoints

plane/api/epics.py

@@ -1,6 +1,16 @@
+from __future__ import annotations
+
 from typing import Any
 
-from ..models.epics import Epic, PaginatedEpicResponse
+from ..models.epics import (
+    AddEpicWorkItems,
+    CreateEpic,
+    Epic,
+    EpicIssue,
+    PaginatedEpicIssueResponse,
+    PaginatedEpicResponse,
+    UpdateEpic,
+)
 from ..models.query_params import PaginatedQueryParams, RetrieveQueryParams
 from .base_resource import BaseResource
 
@@ -9,6 +19,69 @@ class Epics(BaseResource):
     def __init__(self, config: Any) -> None:
         super().__init__(config, "/workspaces/")
 
+    def create(self, workspace_slug: str, project_id: str, data: CreateEpic) -> Epic:
+        """Create a new epic in a project.
+
+        Args:
+            workspace_slug: The workspace slug identifier
+            project_id: UUID of the project
+            data: Epic creation data
+        """
+        # enable epics feature flag
+        response = self._post(
+            f"{workspace_slug}/projects/{project_id}/epics",
+            data.model_dump(exclude_none=True),
+        )
+        return Epic.model_validate(response)
+
+    def retrieve(
+        self,
+        workspace_slug: str,
+        project_id: str,
+        epic_id: str,
+        params: RetrieveQueryParams | None = None,
+    ) -> Epic:
+        """Retrieve an epic by ID.
+
+        Args:
+            workspace_slug: The workspace slug identifier
+            project_id: UUID of the project
+            epic_id: UUID of the epic
+            params: Optional query parameters for expand, fields, etc.
+        """
+        query_params = params.model_dump(exclude_none=True) if params else None
+        response = self._get(
+            f"{workspace_slug}/projects/{project_id}/epics/{epic_id}", params=query_params
+        )
+        return Epic.model_validate(response)
+
+    def update(
+        self, workspace_slug: str, project_id: str, epic_id: str, data: UpdateEpic
+    ) -> Epic:
+        """Partially update an existing epic.
+
+        Args:
+            workspace_slug: The workspace slug identifier
+            project_id: UUID of the project
+            epic_id: UUID of the epic
+            data: Epic update data
+        """
+        response = self._patch(
+            f"{workspace_slug}/projects/{project_id}/epics/{epic_id}",
+            data.model_dump(exclude_none=True),
+        )
+        return Epic.model_validate(response)
+
+    def delete(self, workspace_slug: str, project_id: str, epic_id: str) -> None:
+        """Delete an epic.
+
+        Args:
+            workspace_slug: The workspace slug identifier
+            project_id: UUID of the project
+            epic_id: UUID of the epic
+        """
+        self._delete(f"{workspace_slug}/projects/{project_id}/epics/{epic_id}")
+
     def list(
         self,
         workspace_slug: str,
@@ -26,23 +99,45 @@ def list(
         response = self._get(f"{workspace_slug}/projects/{project_id}/epics", params=query_params)
         return PaginatedEpicResponse.model_validate(response)
 
-    def retrieve(
+    def list_issues(
         self,
         workspace_slug: str,
         project_id: str,
         epic_id: str,
-        params: RetrieveQueryParams | None = None,
-    ) -> Epic:
-        """Retrieve an epic by ID.
+        params: PaginatedQueryParams | None = None,
+    ) -> PaginatedEpicIssueResponse:
+        """List work items under an epic.
 
         Args:
             workspace_slug: The workspace slug identifier
             project_id: UUID of the project
             epic_id: UUID of the epic
-            params: Optional query parameters for expand, fields, etc.
+            params: Optional query parameters for pagination
         """
         query_params = params.model_dump(exclude_none=True) if params else None
         response = self._get(
-            f"{workspace_slug}/projects/{project_id}/epics/{epic_id}", params=query_params
+            f"{workspace_slug}/projects/{project_id}/epics/{epic_id}/issues",
+            params=query_params,
         )
-        return Epic.model_validate(response)
+        return PaginatedEpicIssueResponse.model_validate(response)
+
+    def add_issues(
+        self,
+        workspace_slug: str,
+        project_id: str,
+        epic_id: str,
+        data: AddEpicWorkItems,
+    ) -> list[EpicIssue]:
+        """Add work items as sub-issues under an epic.
+
+        Args:
+            workspace_slug: The workspace slug identifier
+            project_id: UUID of the project
+            epic_id: UUID of the epic
+            data: Work item IDs to add
+        """
+        response = self._post(
+            f"{workspace_slug}/projects/{project_id}/epics/{epic_id}/issues",
+            data.model_dump(exclude_none=True),
+        )
+        return [EpicIssue.model_validate(item) for item in response]

plane/models/epics.py

@@ -43,9 +43,101 @@ class Epic(BaseModel):
     labels: list[str] | None = None
 
 
+class CreateEpic(BaseModel):
+    """Request model for creating an epic."""
+
+    model_config = ConfigDict(extra="ignore", populate_by_name=True)
+
+    name: str
+    description_html: str | None = None
+    state_id: str | None = None
+    parent_id: str | None = None
+    assignee_ids: list[str] | None = None
+    label_ids: list[str] | None = None
+    priority: PriorityEnum | None = None
+    start_date: str | None = None
+    target_date: str | None = None
+    estimate_point: str | None = None
+    external_source: str | None = None
+    external_id: str | None = None
+
+
+class UpdateEpic(BaseModel):
+    """Request model for updating an epic."""
+
+    model_config = ConfigDict(extra="ignore", populate_by_name=True)
+
+    name: str | None = None
+    description_html: str | None = None
+    state_id: str | None = None
+    parent_id: str | None = None
+    assignee_ids: list[str] | None = None
+    label_ids: list[str] | None = None
+    priority: PriorityEnum | None = None
+    start_date: str | None = None
+    target_date: str | None = None
+    estimate_point: str | None = None
+    external_source: str | None = None
+    external_id: str | None = None
+
+
+class AddEpicWorkItems(BaseModel):
+    """Request model for adding work items to an epic."""
+
+    model_config = ConfigDict(extra="ignore", populate_by_name=True)
+
+    work_item_ids: list[str]
+
+
+class EpicIssue(BaseModel):
+    """Response model for an epic's work item."""
+
+    model_config = ConfigDict(extra="allow", populate_by_name=True)
+
+    id: str
+    type_id: str | None = None
+    parent: str | None = None
+    created_at: str | None = None
+    updated_at: str | None = None
+    deleted_at: str | None = None
+    point: int | None = None
+    name: str
+    description_html: str | None = None
+    description_stripped: str | None = None
+    description_binary: str | None = None
+    priority: PriorityEnum | None = None
+    start_date: str | None = None
+    target_date: str | None = None
+    sequence_id: int | None = None
+    sort_order: float | None = None
+    completed_at: str | None = None
+    archived_at: str | None = None
+    last_activity_at: str | None = None
+    is_draft: bool | None = None
+    external_source: str | None = None
+    external_id: str | None = None
+    created_by: str | None = None
+    updated_by: str | None = None
+    project: str | None = None
+    workspace: str | None = None
+    state: str | None = None
+    estimate_point: str | None = None
+    type: str | None = None
+    assignees: list[str] | None = None
+    labels: list[str] | None = None
+
+
 class PaginatedEpicResponse(PaginatedResponse):
     """Paginated response for epics."""
 
     model_config = ConfigDict(extra="allow", populate_by_name=True)
 
     results: list[Epic]
+
+
+class PaginatedEpicIssueResponse(PaginatedResponse):
+    """Paginated response for epic issues."""
+
+    model_config = ConfigDict(extra="allow", populate_by_name=True)
+
+    results: list[EpicIssue]

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "plane-sdk"
-version = "0.2.8"
+version = "0.2.9"
 description = "Python SDK for Plane API"
 readme = "README.md"
 requires-python = ">=3.10"

tests/unit/test_epics.py

@@ -1,8 +1,14 @@
 """Unit tests for Epics API resource (smoke tests with real HTTP requests)."""
 
+from datetime import datetime
+
+import pytest
+
 from plane.client import PlaneClient
-from plane.models.projects import Project
+from plane.models.epics import AddEpicWorkItems, CreateEpic, Epic, UpdateEpic
+from plane.models.projects import Project, ProjectFeature
 from plane.models.query_params import PaginatedQueryParams
+from plane.models.work_items import CreateWorkItem
 
 
 class TestEpicsAPI:
@@ -28,3 +34,82 @@ def test_list_epics_with_params(
         assert hasattr(response, "results")
         assert len(response.results) <= 10
 
+
+class TestEpicsAPICRUD:
+    """Test Epics API CRUD operations."""
+
+    @pytest.fixture
+    def epic(
+        self, client: PlaneClient, workspace_slug: str, project: Project
+    ) -> Epic:
+        """Create a test epic and yield it, then delete it."""
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
+        data = CreateEpic(name=f"Test Epic {timestamp}", priority="high")
+        epic = client.epics.create(workspace_slug, project.id, data)
+        yield epic
+        try:
+            client.epics.delete(workspace_slug, project.id, epic.id)
+        except Exception:
+            pass
+
+    def test_create_epic(
+        self, client: PlaneClient, workspace_slug: str, project: Project
+    ) -> None:
+        """Test creating an epic."""
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
+        data = CreateEpic(name=f"Test Epic {timestamp}")
+        client.projects.update_features(workspace_slug, project.id, ProjectFeature(epics=True))
+        epic = client.epics.create(workspace_slug, project.id, data)
+        assert epic is not None
+        assert epic.id is not None
+        assert epic.name == data.name
+
+    def test_retrieve_epic(
+        self, client: PlaneClient, workspace_slug: str, project: Project, epic: Epic
+    ) -> None:
+        """Test retrieving an epic."""
+        retrieved = client.epics.retrieve(workspace_slug, project.id, epic.id)
+        assert retrieved is not None
+        assert retrieved.id == epic.id
+        assert retrieved.name == epic.name
+
+    def test_update_epic(
+        self, client: PlaneClient, workspace_slug: str, project: Project, epic: Epic
+    ) -> None:
+        """Test updating an epic."""
+        update_data = UpdateEpic(name="Updated Epic Name")
+        updated = client.epics.update(workspace_slug, project.id, epic.id, update_data)
+        assert updated is not None
+        assert updated.id == epic.id
+        assert updated.name == "Updated Epic Name"
+
+    def test_list_epic_issues(
+        self, client: PlaneClient, workspace_slug: str, project: Project, epic: Epic
+    ) -> None:
+        """Test listing work items under an epic."""
+        response = client.epics.list_issues(workspace_slug, project.id, epic.id)
+        assert response is not None
+        assert hasattr(response, "results")
+        assert isinstance(response.results, list)
+
+    def test_add_issues_to_epic(
+        self, client: PlaneClient, workspace_slug: str, project: Project, epic: Epic
+    ) -> None:
+        """Test adding work items to an epic."""
+        work_item = client.work_items.create(
+            workspace_slug,
+            project.id,
+            CreateWorkItem(name="Test Work Item for Epic"),
+        )
+        try:
+            data = AddEpicWorkItems(work_item_ids=[work_item.id])
+            added = client.epics.add_issues(workspace_slug, project.id, epic.id, data)
+            assert added is not None
+            assert isinstance(added, list)
+            assert len(added) == 1
+            assert added[0].parent == epic.id
+        finally:
+            try:
+                client.work_items.delete(workspace_slug, project.id, work_item.id)
+            except Exception:
+                pass