Add signed URL support for storage objects

Expose a storage API method to request signed URLs and add coverage for expiry behavior and error handling so SDK users can fetch temporary download links.

Made-with: Cursor

Author: marcos-muino-garcia

tests/test_yepcode_storage.py

@@ -0,0 +1,75 @@
+import time
+from datetime import datetime, timezone
+
+import pytest
+import requests
+
+from yepcode_run import YepCodeStorage
+from yepcode_run.api.yepcode_api import YepCodeApiError
+
+
+TEST_NAME = "test-run-sdk.txt"
+TEST_CONTENT = b"hello signed url"
+
+
+@pytest.fixture
+def storage():
+    return YepCodeStorage()
+
+
+@pytest.fixture
+def uploaded_file(storage):
+    storage.upload(TEST_NAME, TEST_CONTENT)
+    yield TEST_NAME
+    try:
+        storage.delete(TEST_NAME)
+    except Exception:
+        pass
+
+
+def _parse_iso(value: str) -> float:
+    return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
+
+
+@pytest.mark.skip(reason="Requires the signed-urls endpoint deployed in the target environment")
+def test_create_signed_url_default_expiry(storage, uploaded_file):
+    result = storage.create_signed_url(uploaded_file)
+
+    assert isinstance(result.url, str) and len(result.url) > 0
+    assert result.path == uploaded_file
+
+    expires_at = _parse_iso(result.expires_at)
+    expected_expiry = time.time() + 3600
+    assert abs(expires_at - expected_expiry) < 60
+
+
+@pytest.mark.skip(reason="Requires the signed-urls endpoint deployed in the target environment")
+def test_create_signed_url_custom_expiry(storage, uploaded_file):
+    result = storage.create_signed_url(uploaded_file, expires_in_seconds=60)
+
+    expires_at = _parse_iso(result.expires_at)
+    expected_expiry = time.time() + 60
+    assert abs(expires_at - expected_expiry) < 30
+
+
+@pytest.mark.skip(reason="Requires the signed-urls endpoint deployed in the target environment")
+def test_create_signed_url_returns_fetchable_url(storage, uploaded_file):
+    result = storage.create_signed_url(uploaded_file)
+
+    response = requests.get(result.url, timeout=30)
+    assert response.ok
+    assert response.content == TEST_CONTENT
+
+
+@pytest.mark.skip(reason="Requires the signed-urls endpoint deployed in the target environment")
+def test_create_signed_url_missing_file_raises_404(storage):
+    with pytest.raises(YepCodeApiError) as exc_info:
+        storage.create_signed_url("does-not-exist.txt")
+    assert exc_info.value.status == 404
+
+
+@pytest.mark.skip(reason="Requires the signed-urls endpoint deployed in the target environment")
+def test_create_signed_url_out_of_range_expiry_raises_400(storage, uploaded_file):
+    with pytest.raises(YepCodeApiError) as exc_info:
+        storage.create_signed_url(uploaded_file, expires_in_seconds=999999)
+    assert exc_info.value.status == 400

yepcode_run/api/types.py

@@ -470,6 +470,27 @@ class CreateStorageObjectInput:
     file: Any
 
 
+@dataclass
+class CreateSignedUrlInput:
+    path: str
+    expires_in_seconds: Optional[int] = None
+
+
+@dataclass
+class SignedUrl:
+    url: str
+    path: str
+    expires_at: str
+
+    @staticmethod
+    def from_dict(data: dict) -> "SignedUrl":
+        return SignedUrl(
+            url=data["url"],
+            path=data["path"],
+            expires_at=data.get("expiresAt", data.get("expires_at")),
+        )
+
+
 # Dependency manifest types
 @dataclass
 class ProgrammingLanguageManifest:

yepcode_run/api/yepcode_api.py

@@ -42,6 +42,8 @@
     ScheduledProcessInput,
     CreateStorageObjectInput,
     StorageObject,
+    CreateSignedUrlInput,
+    SignedUrl,
     ProgrammingLanguage,
     ProgrammingLanguageManifest,
     UpdateTeamDependenciesInput,
@@ -629,3 +631,10 @@ def delete_object(self, name: str) -> None:
                 response.status_code,
             )
         return None
+
+    def create_signed_url(self, data: CreateSignedUrlInput) -> SignedUrl:
+        body: Dict[str, Any] = {"path": data.path}
+        if data.expires_in_seconds is not None:
+            body["expiresInSeconds"] = data.expires_in_seconds
+        response = self._request("POST", "/storage/signed-urls", {"data": body})
+        return SignedUrl.from_dict(response)

yepcode_run/storage/yepcode_storage.py

@@ -1,7 +1,13 @@
 from typing import List, Optional, Dict, Any
 
 from ..api.api_manager import YepCodeApiManager
-from ..api.types import CreateStorageObjectInput, StorageObject, YepCodeApiConfig
+from ..api.types import (
+    CreateSignedUrlInput,
+    CreateStorageObjectInput,
+    SignedUrl,
+    StorageObject,
+    YepCodeApiConfig,
+)
 
 
 class YepCodeStorage:
@@ -27,3 +33,10 @@ def delete(self, name: str) -> None:
 
     def list(self, **kwargs) -> List[StorageObject]:
         return self._api.get_objects(kwargs if kwargs else None)
+
+    def create_signed_url(
+        self, name: str, expires_in_seconds: Optional[int] = None
+    ) -> SignedUrl:
+        return self._api.create_signed_url(
+            CreateSignedUrlInput(path=name, expires_in_seconds=expires_in_seconds)
+        )