diff --git a/plane/api/work_items/attachments.py b/plane/api/work_items/attachments.py index 53bf7a8..046adda 100644 --- a/plane/api/work_items/attachments.py +++ b/plane/api/work_items/attachments.py @@ -1,6 +1,10 @@ +from __future__ import annotations + from collections.abc import Mapping from typing import Any +import requests as _requests + from ...models.work_items import ( UpdateWorkItemAttachment, WorkItemAttachment, @@ -96,6 +100,92 @@ def update( ) return WorkItemAttachment.model_validate(response) + def upload_from_bytes( + self, + workspace_slug: str, + project_id: str, + work_item_id: str, + file_bytes: bytes, + name: str, + content_type: str, + ) -> WorkItemAttachment: + """Upload a file to a work item as an attachment. + + Handles the full three-step flow: + 1. Create the attachment record and receive a presigned S3 upload URL. + 2. Upload the file bytes directly to S3 via the presigned POST. + 3. Mark the attachment as uploaded (PATCH is_uploaded=True). + + Args: + workspace_slug: The workspace slug identifier + project_id: UUID of the project + work_item_id: UUID of the work item + file_bytes: Raw file bytes to upload + name: Filename (e.g. "report.pdf") + content_type: MIME type (e.g. "application/pdf") + """ + size = len(file_bytes) + + # Step 1 — create attachment record, get presigned S3 POST URL + raw = self._post( + f"{workspace_slug}/projects/{project_id}/work-items/{work_item_id}/attachments", + {"name": name, "type": content_type, "size": size}, + ) + upload_data = raw["upload_data"] + asset_id = raw["asset_id"] + attachment = WorkItemAttachment.model_validate(raw["attachment"]) + + # Step 2 — upload bytes to S3 (raw requests, no Plane auth headers) + fields = upload_data.get("fields", {}) + s3_resp = _requests.post( + upload_data["url"], + data=fields, + files={"file": (name, file_bytes, content_type)}, + timeout=120, + ) + s3_resp.raise_for_status() + + # Step 3 — mark as uploaded (returns 204, no body) + self._patch( + f"{workspace_slug}/projects/{project_id}/work-items/{work_item_id}/attachments/{asset_id}", + {"is_uploaded": True}, + ) + + return attachment + + def get_download_url( + self, + workspace_slug: str, + project_id: str, + work_item_id: str, + attachment_id: str, + ) -> str: + """Get a presigned download URL for a work item attachment. + + Calls the attachment detail endpoint which issues a redirect to a + time-limited presigned S3 URL. The returned URL can be opened in a + browser or fetched with any HTTP client — no Plane auth required. + + 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 + + Returns: + Presigned S3 URL (time-limited, typically ~1 hour) + """ + url = self._build_url( + f"{workspace_slug}/projects/{project_id}/work-items/{work_item_id}/attachments/{attachment_id}" + ) + resp = self.session.get(url, headers=self._headers(), allow_redirects=False, timeout=self.config.timeout) + if resp.status_code in (301, 302, 303, 307, 308): + location = resp.headers.get("Location") + if location: + return location + # Not a redirect — let _handle_response raise or return + return self._handle_response(resp) + def delete( self, workspace_slug: str, project_id: str, work_item_id: str, attachment_id: str ) -> None: diff --git a/pyproject.toml b/pyproject.toml index 10eec4e..18218a5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "plane-sdk" -version = "0.2.15" +version = "0.2.16" description = "Python SDK for Plane API" readme = "README.md" requires-python = ">=3.10"