Add artifact downloads and asset uploads

Author: rsgalloway

README.md

@@ -120,7 +120,8 @@ remote state before repeating a create, publish, or execute request.
 
 ## Documentation
 
-See the [documentation](docs/index.md) for installation, Python usage, CLI options,
+See the [API and file workflow guide](docs/api.md) for asset uploads and artifact
+downloads, and the [documentation](docs/index.md) for installation, Python usage, CLI options,
 and examples. Browse [Subfork Examples](https://examples.subfork.com) and the
 [subfork-examples repository](https://github.com/subforkdev/subfork-examples) for
 graphs to learn from and reuse.

docs/api.md

@@ -0,0 +1,119 @@
+# API reference and file workflows
+
+The synchronous Python client returns dictionaries from the Subfork API. It does
+not print results; the command-line tool's output options do not affect Python.
+Set `SUBFORK_API_KEY`, then use a context manager to close connections reliably.
+
+## Execute and inspect results
+
+```python
+from typing import Any, Dict
+from subfork import Subfork
+
+
+def run_graph(client: Subfork, graph_id: str) -> Dict[str, Any]:
+    """Run a graph and return its completed outputs."""
+    submitted = client.graphs.execute(graph_id)
+    result = client.executions.wait(submitted["id"], timeout=300)
+    if result["status"] != "completed":
+        raise RuntimeError(f"Execution {submitted['id']}: {result['status']}")
+    return result["outputs"]
+
+
+with Subfork() as client:
+    outputs = run_graph(client, "g_YOUR_GRAPH")
+```
+
+Output names and shapes are defined by each graph. Text and JSON can be used
+immediately. A media output may contain an `artifact` object, a remote URL, or
+inline data. Artifact download helpers accept an artifact ID, not arbitrary URLs.
+Waiting requires `graphs:read`; starting a run requires `graphs:run`. A waiting
+timeout does not cancel the remote execution.
+
+## Download artifacts
+
+For a graph exposing an `audio` media output with an artifact reference:
+
+```python
+with Subfork() as client:
+    outputs = run_graph(client, "g_YOUR_GRAPH")
+    artifact_id = outputs["audio"]["artifact"]["artifact_id"]
+    path = client.artifacts.download(artifact_id, "narration.mp3")
+```
+
+A direct artifact output instead exposes `outputs["artifact"]["artifact_id"]`.
+The helper works for PDFs, images, MP3, MP4 and other stored file types. It returns
+`pathlib.Path`; it does not decode media or open a viewer. Use your preferred media
+library or player after downloading.
+
+`download(artifact_id, destination, *, overwrite=False, max_bytes=250_000_000)`
+streams to a temporary file, then publishes the complete file at the requested
+path. The parent directory must exist. Existing files are protected unless you
+pass `overwrite=True`; failed downloads preserve them. The default decoded-byte
+limit is 250 MB and can be increased explicitly.
+
+Downloads require `graphs:read` and access to the artifact. Expired or deleted
+artifacts cannot be recovered by the client. One HTTPS storage redirect is
+supported; the Subfork authorization header and cookies are not forwarded.
+
+## Upload graph assets
+
+**Server requirement:** API-key asset upload/list support must be deployed.
+Older servers return HTTP 403 even with an otherwise valid key.
+
+```python
+with Subfork() as client:
+    uploaded = client.assets.upload(
+        "g_YOUR_GRAPH", "document.pdf", content_type="application/pdf"
+    )
+    artifact_id = uploaded["artifact_id"]
+    print(artifact_id)
+```
+
+`upload(graph_id, source, *, content_type=None)` streams a local file as multipart
+form data. The MIME type is inferred from its filename unless supplied; the
+server validates the file and controls persisted metadata. The returned JSON
+includes `artifact_id`, `media_type`, `size_bytes` and other artifact metadata.
+
+Uploads require `graphs:write`, an owned graph, enabled uploads and available
+storage quota. The asset starts private. Uploading does **not** automatically
+bind an Asset node or execute the graph. To select it on an existing Asset node:
+
+```python
+with Subfork() as client:
+    graph_id = "g_YOUR_GRAPH"
+    graph = client.graphs.get(graph_id)
+    definition = graph["definition"]
+    source_node = next(
+        node for node in definition["nodes"]
+        if node["node_instance_id"] == "pdf" and node["node_id"] == "n_asset"
+    )
+    uploaded = client.assets.upload(graph_id, "document.pdf")
+    source_node["params"]["artifact_id"] = uploaded["artifact_id"]
+    client.graphs.update(
+        graph_id, name=graph["name"], definition=definition,
+        description=graph.get("description") or "",
+    )
+```
+
+The example changes the draft; existing published versions remain immutable.
+Fetch the latest draft and avoid concurrent edits when updating its definition.
+For a graph exposing an artifact input, you can instead supply the reference in
+`graphs.execute(..., inputs={"pdf": {"artifact_id": artifact_id}})`. The input
+name must match that graph's published or draft interface.
+
+List assets with `client.assets.list(graph_id)`. It returns an `assets` array in a
+JSON object. Use `include_generated=True` to include unexpired execution outputs.
+Listing requires `graphs:read` and ownership of the graph.
+
+The client does not expose secret management, asset deletion, visibility changes
+or retention changes. Configure provider secrets through the Subfork UI.
+
+## Errors and retries
+
+HTTP failures raise `APIError` subclasses, including permission, validation and
+rate-limit errors. Downloads may also raise `FileExistsError`, filesystem errors,
+or `ValueError` for an invalid limit or oversized response. Network failures raise
+`TransportError`. File transfers and graph executions are never automatically
+retried: an upload or run may have succeeded before the connection failed. Inspect
+remote state before repeating a write.

docs/mkpages.yml

@@ -8,6 +8,8 @@ navigation:
     href: /installation/
   - label: Python
     href: /usage/
+  - label: API
+    href: /api/
   - label: CLI
     href: /cli/
   - label: Examples

docs/usage.md

@@ -87,3 +87,6 @@ requests. After a `TransportError`, check remote state before submitting another
 create, publish, or execute request.
 
 See [troubleshooting](troubleshooting.md) for common errors.
+
+See the [API and file workflow guide](api.md) for asset uploads, artifact downloads,
+and using execution results in your own functions.

src/subfork/client.py

@@ -5,6 +5,8 @@
 import math
 import os
 import re
+import tempfile
+from pathlib import Path
 from types import TracebackType
 from typing import Any, Optional, Type
 
@@ -20,7 +22,7 @@
     TransportError,
     ValidationError,
 )
-from .resources import Executions, Graphs, Nodes
+from .resources import Artifacts, Assets, Executions, Graphs, Nodes
 
 
 def _error_message(response: httpx.Response) -> str:
@@ -30,7 +32,7 @@ def _error_message(response: httpx.Response) -> str:
         return message
     try:
         payload = response.json()
-    except ValueError:
+    except (ValueError, httpx.ResponseNotRead):
         return message
     detail = payload.get("detail") if isinstance(payload, dict) else None
     if isinstance(detail, str) and re.fullmatch(
@@ -107,6 +109,8 @@ def __init__(
             follow_redirects=False,
             transport=transport,
         )
+        self.artifacts = Artifacts(self)
+        self.assets = Assets(self)
         self.nodes = Nodes(self)
         self.graphs = Graphs(self)
         self.executions = Executions(self)
@@ -124,6 +128,17 @@ def _request(self, method: str, path: str, **kwargs: Any) -> Any:
             raise TransportError(
                 "API request could not be completed; inspect remote state before retrying a write."
             ) from None
+        self._check_response(response)
+        if response.status_code == 204:
+            return None
+        try:
+            return response.json()
+        except ValueError:
+            raise InvalidResponseError("API returned invalid JSON.") from None
+
+    @staticmethod
+    def _check_response(response: httpx.Response) -> None:
+        """Raise a credential-safe API exception for an unsuccessful response."""
         if not response.is_success:
             errors = {
                 401: AuthenticationError,
@@ -138,12 +153,61 @@ def _request(self, method: str, path: str, **kwargs: Any) -> Any:
                 status_code=response.status_code,
                 retry_after=response.headers.get("retry-after"),
             )
-        if response.status_code == 204:
-            return None
+
+    def _download(self, path: str, destination: Path, *, overwrite: bool, max_bytes: int) -> Path:
+        """Stream an artifact, stripping credentials on a storage redirect."""
+        if max_bytes <= 0:
+            raise ValueError("max_bytes must be positive.")
+        if destination.exists() and not overwrite:
+            raise FileExistsError("Output file already exists; set overwrite=True.")
+        temporary = None
+        response = None
         try:
-            return response.json()
-        except ValueError:
-            raise InvalidResponseError("API returned invalid JSON.") from None
+            request = self._client.build_request("GET", path.lstrip("/"))
+            response = self._client.send(request, stream=True)
+            if response.status_code in {301, 302, 303, 307, 308}:
+                location = response.headers.get("location")
+                if not location:
+                    raise InvalidResponseError("Artifact redirect has no location.")
+                try:
+                    target = response.url.join(location)
+                except httpx.InvalidURL:
+                    raise InvalidResponseError(
+                        "Artifact redirect contains an invalid URL."
+                    ) from None
+                if target.scheme != "https" or target.userinfo or target.fragment:
+                    raise InvalidResponseError(
+                        "Artifact redirect must use HTTPS without credentials."
+                    )
+                response.close()
+                # A fresh Request does not inherit the API client's headers or cookies.
+                response = self._client.send(
+                    httpx.Request("GET", target), stream=True, auth=None, follow_redirects=False
+                )
+            self._check_response(response)
+            with tempfile.NamedTemporaryFile(
+                dir=destination.parent, prefix=".subfork-", delete=False
+            ) as stream:
+                temporary = Path(stream.name)
+                size = 0
+                for chunk in response.iter_bytes(chunk_size=65536):
+                    size += len(chunk)
+                    if size > max_bytes:
+                        raise ValueError("Artifact exceeds max_bytes; partial download discarded.")
+                    stream.write(chunk)
+            if overwrite:
+                os.replace(temporary, destination)
+            else:
+                # Exclusive publication also protects against another writer racing us.
+                os.link(temporary, destination)
+            return destination
+        except httpx.TransportError:
+            raise TransportError("Artifact download could not be completed.") from None
+        finally:
+            if response is not None:
+                response.close()
+            if temporary is not None:
+                temporary.unlink(missing_ok=True)
 
     def close(self) -> None:
         """Release the HTTP connection pool."""

src/subfork/resources.py

@@ -3,8 +3,11 @@
 from __future__ import annotations
 
 import math
+import mimetypes
+import os
 import time
-from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Union
 from urllib.parse import quote
 
 from .errors import ExecutionTimeout
@@ -226,3 +229,67 @@ def wait(
             }:
                 return result
             time.sleep(min(poll_interval, max(0, deadline - time.monotonic())))
+
+
+class Artifacts(Resource):
+    """Download artifacts accessible to the API-key account."""
+
+    def download(
+        self,
+        artifact_id: str,
+        destination: Union[str, os.PathLike],
+        *,
+        overwrite: bool = False,
+        max_bytes: int = 250_000_000,
+    ) -> Path:
+        """Stream bytes to a file and return its path after successful completion.
+
+        The parent directory must exist. Failed downloads leave no partial output
+        and preserve an existing destination. Set overwrite explicitly to replace
+        it. At most one HTTPS storage redirect is followed, without API credentials.
+        The default decoded-byte limit is 250 MB; increase it for larger media.
+        Requires graphs:read and access to the unexpired artifact.
+        """
+        return self._client._download(
+            "/artifacts/" + segment(artifact_id),
+            Path(destination),
+            overwrite=overwrite,
+            max_bytes=max_bytes,
+        )
+
+
+class Assets(Resource):
+    """Manage input files belonging to an owned graph."""
+
+    def list(self, graph_id: str, *, include_generated: bool = False) -> Dict[str, Any]:
+        """Return the server's assets envelope; generated outputs are opt-in."""
+        return self._client._request(
+            "GET",
+            "/graphs/" + segment(graph_id) + "/assets",
+            params={"include_generated": include_generated},
+        )
+
+    def upload(
+        self,
+        graph_id: str,
+        source: Union[str, os.PathLike],
+        *,
+        content_type: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Upload a local file and return artifact metadata, including artifact_id.
+
+        Requires graphs:write on a server supporting API-key asset uploads.
+        Upload quotas, file validation and the uploads feature switch still apply.
+        This stores an asset; it does not select it on a node or execute the graph.
+        The file is streamed from disk and the operation is never retried.
+        """
+        path = Path(source)
+        media_type = (
+            content_type or 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) + "/assets/files",
+                files={"upload": (path.name, stream, media_type)},
+            )