Skip to content
Open
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
44 changes: 38 additions & 6 deletions plane/api/work_items/attachments.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
from ...models.work_items import (
UpdateWorkItemAttachment,
WorkItemAttachment,
WorkItemAttachmentCreateResponse,
WorkItemAttachmentUploadRequest,
)
from ..base_resource import BaseResource
Expand Down Expand Up @@ -58,20 +59,30 @@ def create(
project_id: str,
work_item_id: str,
data: WorkItemAttachmentUploadRequest,
) -> WorkItemAttachment:
) -> WorkItemAttachmentCreateResponse:
"""Create an attachment for a work item.

Plane returns a wrapper containing both the created attachment record
and an S3 multipart-POST policy in ``upload_data``. The caller posts
the file as ``multipart/form-data`` to ``upload_data["url"]`` with the
``upload_data["fields"]`` plus a ``file`` part, then calls ``update``
with ``is_uploaded=True`` to mark the attachment ready.

Args:
workspace_slug: The workspace slug identifier
project_id: UUID of the project
work_item_id: UUID of the work item
data: Attachment data
data: Attachment data (filename, size, MIME type, etc.)

Returns:
WorkItemAttachmentCreateResponse with ``attachment`` record and
``upload_data`` S3 policy.
"""
response = self._post(
f"{workspace_slug}/projects/{project_id}/work-items/{work_item_id}/attachments",
data.model_dump(exclude_none=True),
)
return WorkItemAttachment.model_validate(response)
return WorkItemAttachmentCreateResponse.model_validate(response)

def update(
self,
Expand All @@ -83,18 +94,39 @@ def update(
) -> WorkItemAttachment:
"""Update an attachment for a work item.

The Plane API responds to attachment PATCH with ``204 No Content``
and exposes no metadata-by-id endpoint (GET on a single attachment
URL serves the file via S3 redirect). To return the updated record
per CRUD convention, this method follows the PATCH with a ``list``
call and filters by id.

Args:
workspace_slug: The workspace slug identifier
project_id: UUID of the project
work_item_id: UUID of the work item
attachment_id: UUID of the attachment
data: Updated attachment data

Returns:
Updated WorkItemAttachment record.

Raises:
ValueError: If the attachment cannot be found after the update.
Plane only includes attachments with ``is_uploaded=True``
in list responses, so updating an unuploaded attachment to
stay unuploaded will raise.
"""
response = self._patch(
f"{workspace_slug}/projects/{project_id}/work-items/{work_item_id}/attachments/{attachment_id}",
self._patch(
f"{workspace_slug}/projects/{project_id}/work-items/{work_item_id}/attachments/{attachment_id}/",
data.model_dump(exclude_none=True),
)
return WorkItemAttachment.model_validate(response)
for attachment in self.list(workspace_slug, project_id, work_item_id):
if attachment.id == attachment_id:
return attachment
raise ValueError(
f"Attachment {attachment_id} not found after update; "
"Plane only lists attachments with is_uploaded=True."
)

def delete(
self, workspace_slug: str, project_id: str, work_item_id: str, attachment_id: str
Expand Down
17 changes: 17 additions & 0 deletions plane/models/work_items.py
Original file line number Diff line number Diff line change
Expand Up @@ -438,6 +438,23 @@ class UpdateWorkItemAttachment(BaseModel):
)


class WorkItemAttachmentCreateResponse(BaseModel):
"""Response from creating a work item attachment.

Plane returns a wrapper containing both the created attachment record and
the S3 multipart-POST policy needed to upload the file bytes. After the
upload completes, the caller should call ``WorkItemAttachments.update``
with ``is_uploaded=True`` to mark the attachment ready.
"""

model_config = ConfigDict(extra="allow", populate_by_name=True)

attachment: WorkItemAttachment
upload_data: dict[str, Any]
asset_id: str | None = None
asset_url: str | None = None


class WorkItemRelation(BaseModel):
"""Work item relation model."""

Expand Down
96 changes: 96 additions & 0 deletions tests/unit/test_work_item_attachments.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,96 @@
"""Unit tests for WorkItemAttachments API resource (smoke tests with real HTTP requests)."""

from collections.abc import Iterator
from contextlib import suppress

import pytest

from plane.client import PlaneClient
from plane.models.projects import Project
from plane.models.work_items import (
CreateWorkItem,
UpdateWorkItemAttachment,
WorkItemAttachmentCreateResponse,
WorkItemAttachmentUploadRequest,
)


class TestWorkItemAttachmentsAPI:
"""Test WorkItemAttachments API resource."""

@pytest.fixture
def work_item_id(
self, client: PlaneClient, workspace_slug: str, project: Project
) -> Iterator[str]:
"""Create a work item for attachment tests and clean it up after."""
work_item = client.work_items.create(
workspace_slug,
project.id,
CreateWorkItem(name="Attachment test work item"),
)
yield work_item.id
with suppress(Exception):
client.work_items.delete(workspace_slug, project.id, work_item.id)

def test_list_empty(
self, client: PlaneClient, workspace_slug: str, project: Project, work_item_id: str
) -> None:
"""Newly created work item should have no attachments."""
attachments = client.work_items.attachments.list(workspace_slug, project.id, work_item_id)
assert attachments == []

def test_create_returns_wrapper(
self, client: PlaneClient, workspace_slug: str, project: Project, work_item_id: str
) -> None:
"""Create should return wrapper with both attachment record and S3 upload policy."""
result = client.work_items.attachments.create(
workspace_slug,
project.id,
work_item_id,
WorkItemAttachmentUploadRequest(name="test.txt", size=12, type="text/plain"),
)
try:
assert isinstance(result, WorkItemAttachmentCreateResponse)
assert result.attachment.id is not None
assert result.attachment.asset
assert not result.attachment.is_uploaded
assert "url" in result.upload_data
assert "fields" in result.upload_data
finally:
with suppress(Exception):
client.work_items.attachments.delete(
workspace_slug, project.id, work_item_id, result.attachment.id
)

def test_full_lifecycle(
self, client: PlaneClient, workspace_slug: str, project: Project, work_item_id: str
) -> None:
"""Create -> mark uploaded -> list -> delete."""
created = client.work_items.attachments.create(
workspace_slug,
project.id,
work_item_id,
WorkItemAttachmentUploadRequest(name="lifecycle.txt", size=20, type="text/plain"),
)
attachment_id = created.attachment.id
assert attachment_id is not None

try:
# Plane filters list to is_uploaded=True only — mark uploaded first.
updated = client.work_items.attachments.update(
workspace_slug,
project.id,
work_item_id,
attachment_id,
UpdateWorkItemAttachment(is_uploaded=True),
)
assert updated.id == attachment_id
assert updated.is_uploaded is True

listed = client.work_items.attachments.list(workspace_slug, project.id, work_item_id)
assert any(a.id == attachment_id for a in listed)
finally:
with suppress(Exception):
client.work_items.attachments.delete(
workspace_slug, project.id, work_item_id, attachment_id
)