Add graph media uploads to the Python client

Author: rsgalloway

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "subfork"
-version = "2.0.1"
+version = "2.0.2"
 description = "Python API client for Subfork graph authoring and execution"
 readme = "README.md"
 license = {file = "LICENSE"}

src/subfork/resources.py

@@ -95,6 +95,33 @@ def create(
             },
         )
 
+    def upload_media(
+        self,
+        graph_id: str,
+        purpose: str,
+        file_path: Union[str, Path],
+        *,
+        replace_id: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Upload a thumbnail, preview, or screenshot to an owned graph.
+
+        Requires uploads. Graph media is stored by the server and counts toward
+        graph media quotas. The request is not automatically retried.
+        """
+        if purpose not in {"thumbnail", "preview", "screenshot"}:
+            raise ValueError("purpose must be thumbnail, preview, or screenshot.")
+        if replace_id is not None and (purpose != "screenshot" or not replace_id):
+            raise ValueError("replace_id is supported only for screenshots.")
+        path = Path(file_path)
+        media_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
+        with path.open("rb") as stream:
+            return self._client._request(
+                "POST",
+                "/graphs/" + segment(graph_id) + "/media/" + purpose,
+                params={"replace_id": replace_id} if replace_id else None,
+                files={"upload": (path.name, stream, media_type)},
+            )
+
     def update(
         self, graph_id: str, *, name: str, definition: Dict[str, Any], description: str = ""
     ) -> Dict[str, Any]:

tests/test_client.py

@@ -1,6 +1,7 @@
 """Exercise client contracts without network access or real credentials."""
 
 import json
+from pathlib import Path
 from typing import Any, Type
 
 import httpx
@@ -141,6 +142,37 @@ def handle(request: httpx.Request) -> httpx.Response:
     assert seen[-1][2:] == ({"version": "v1"}, {"inputs": {"value": 3}})
 
 
+def test_graph_media_upload_uses_multipart(tmp_path: Path) -> None:
+    """Upload graph media as a typed multipart file without retrying."""
+    image = tmp_path / "thumbnail.jpg"
+    image.write_bytes(b"jpeg-bytes")
+    seen = []
+
+    def handle(request: httpx.Request) -> httpx.Response:
+        seen.append(request)
+        return httpx.Response(201, json={"artifact_id": "art_thumb"})
+
+    with Subfork(KEY, transport=httpx.MockTransport(handle)) as client:
+        result = client.graphs.upload_media("g_test", "thumbnail", image)
+        replacement = client.graphs.upload_media(
+            "g_test", "screenshot", image, replace_id="art_old"
+        )
+        with pytest.raises(ValueError, match="purpose"):
+            client.graphs.upload_media("g_test", "unknown", image)
+
+    assert result == replacement == {"artifact_id": "art_thumb"}
+    assert len(seen) == 2
+    assert seen[0].method == "POST"
+    assert seen[0].url.path == "/api/v1/graphs/g_test/media/thumbnail"
+    assert str(seen[0].url.params) == ""
+    assert seen[1].url.path == "/api/v1/graphs/g_test/media/screenshot"
+    assert str(seen[1].url.params) == "replace_id=art_old"
+    assert seen[0].headers["content-type"].startswith("multipart/form-data;")
+    assert b'filename="thumbnail.jpg"' in seen[0].content
+    assert b"image/jpeg" in seen[0].content
+    assert b"jpeg-bytes" in seen[0].content
+
+
 def test_transport_failure_is_not_retried() -> None:
     """Verify transport failure is not retried."""
     seen = []