Skip to content
Draft
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
90 changes: 90 additions & 0 deletions plane/api/work_items/attachments.py
Original file line number Diff line number Diff line change
@@ -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,
Expand Down Expand Up @@ -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:
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down