diff --git a/README.md b/README.md index fada157..cccbd4c 100644 --- a/README.md +++ b/README.md @@ -47,8 +47,8 @@ subfork execute GRAPH_ID --version v1 subfork execute GRAPH_ID -o results.json ``` -`execute` waits for completion and returns JSON. Use `-o` to save results to a -file; progress stays on stderr. Add `-f` / `--force` to overwrite existing output +`execute` waits for completion and shows progress on stderr. Results are opt-in: +use `-o results.json` to save JSON or `-o -` to print it to stdout. Add `-f` / `--force` to overwrite existing output files with `execute` or `export`. Run `subfork --help` or `subfork execute --help` for more options. @@ -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. diff --git a/docs/api.md b/docs/api.md new file mode 100644 index 0000000..420ed33 --- /dev/null +++ b/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. diff --git a/docs/cli.md b/docs/cli.md index 077144e..7503a11 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -38,21 +38,21 @@ subfork execute GRAPH_ID -o results.json subfork execute GRAPH_ID -o results.json --force ``` -Execution waits for completion and prints JSON outputs. `-o` / `--out` writes -results to a file instead of stdout. Use `-f` / `--force` to overwrite an existing +Execution waits for completion without printing results. `-o` / `--out` writes +result JSON to a file; use `-o -` to print it to stdout. Use `-f` / `--force` to overwrite an existing file; this flag also works with `export`. A yellow spinner shows active nodes. Finished nodes remain on stderr with their -final status. JSON results stay on stdout, so piping works normally: +final status. Opt in to JSON output for piping: ```bash -subfork execute GRAPH_ID > results.json +subfork execute GRAPH_ID -o - > results.json ``` | Option | Behavior | | --- | --- | -| `--no-wait` | Return the submission ID and status immediately | -| `--raw` | Return the full execution snapshot | +| `--no-wait` | Report submission ID and status on stderr immediately; use `-o` for JSON | +| `--raw` | Print the full execution snapshot, or write it to the `-o` destination | | `--wait-timeout 300` | Wait up to 300 seconds; the default is 120 | | `--poll-interval 1` | Poll every second; the default is 2 | diff --git a/docs/mkpages.yml b/docs/mkpages.yml index 1e70b59..2db3121 100644 --- a/docs/mkpages.yml +++ b/docs/mkpages.yml @@ -8,6 +8,8 @@ navigation: href: /installation/ - label: Python href: /usage/ + - label: API + href: /api/ - label: CLI href: /cli/ - label: Examples diff --git a/docs/usage.md b/docs/usage.md index c62bdef..9fc4bed 100644 --- a/docs/usage.md +++ b/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. diff --git a/src/subfork/cli.py b/src/subfork/cli.py index e982bc0..3f36629 100644 --- a/src/subfork/cli.py +++ b/src/subfork/cli.py @@ -183,7 +183,11 @@ def build_parser() -> argparse.ArgumentParser: child.add_argument("--comment", default="") elif command == "execute": child.add_argument("--version", default="draft") - child.add_argument("-o", "--out", help="Write result JSON to a file instead of stdout") + child.add_argument( + "-o", + "--out", + help="Write result JSON to a file, or - for stdout (default: no results)", + ) child.add_argument( "--no-wait", action="store_true", help="Return submission status immediately" ) @@ -314,7 +318,12 @@ def main(argv: Optional[Sequence[str]] = None) -> int: if result_file is not None else (export_file if export_file != "-" else None) ) - if destination is not None and Path(destination).exists() and not force: + if ( + destination is not None + and destination != "-" + and Path(destination).exists() + and not force + ): raise ValueError("Output file already exists; use --force to overwrite.") with Subfork(base_url=args.base_url, timeout=args.timeout) as client: result = graph_command(client, args) @@ -333,16 +342,25 @@ def main(argv: Optional[Sequence[str]] = None) -> int: } else: display = result.get("outputs", {}) - rendered = json.dumps(display, indent=2, ensure_ascii=False) + "\n" - output = getattr(args, "output", "-") - if result_file is None and output == "-": - sys.stdout.write(rendered) - else: - # Open only after the request and serialization succeed. - with Path(result_file if result_file is not None else output).open( - "w" if force else "x", encoding="utf-8", newline="\n" - ) as stream: - stream.write(rendered) + emit_results = args.command != "execute" or result_file is not None or args.raw + if emit_results: + rendered = json.dumps(display, indent=2, ensure_ascii=False) + "\n" + output = result_file if result_file is not None else getattr(args, "output", "-") + if output == "-": + sys.stdout.write(rendered) + else: + # Open only after the request and serialization succeed. + with Path(output).open( + "w" if force else "x", encoding="utf-8", newline="\n" + ) as stream: + stream.write(rendered) + elif args.command == "execute" and args.no_wait: + print( + "Execution {}: {}".format( + result.get("id") or result.get("execution_id"), result.get("status") + ), + file=sys.stderr, + ) if failed: print( "subfork: execution did not complete successfully; use --raw for details", diff --git a/src/subfork/client.py b/src/subfork/client.py index d3209a3..9b397bc 100644 --- a/src/subfork/client.py +++ b/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.""" diff --git a/src/subfork/resources.py b/src/subfork/resources.py index 4cd1349..d8d768f 100644 --- a/src/subfork/resources.py +++ b/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)}, + ) diff --git a/tests/test_cli.py b/tests/test_cli.py index 58dd78e..232e3ec 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -126,7 +126,7 @@ def test_removed_groups(command: str) -> None: assert result.value.code == 2 -@pytest.mark.parametrize("mode", ["default", "raw", "no-wait", "failed", "timeout"]) +@pytest.mark.parametrize("mode", ["default", "stdout", "raw", "no-wait", "failed", "timeout"]) def test_execution_output_modes(monkeypatch: pytest.MonkeyPatch, capsys: Any, mode: str) -> None: """Wait for results by default and preserve explicit diagnostic output modes.""" seen = [] @@ -164,6 +164,8 @@ def timeout(*args: Any, **kwargs: Any) -> dict: arguments = ["execute", "g_test"] if mode in {"raw", "no-wait"}: arguments.append("--" + mode) + if mode == "stdout": + arguments.extend(["-o", "-"]) assert cli.main(arguments) == (1 if mode in {"failed", "timeout"} else 0) captured = capsys.readouterr() assert sum(request.method == "POST" for request in seen) == 1 @@ -171,16 +173,15 @@ def timeout(*args: Any, **kwargs: Any) -> dict: assert captured.out == "" assert "e_test" in captured.err and "not canceled" in captured.err return - output = json.loads(captured.out) - if mode == "raw": - assert output == completed - elif mode == "default": - assert output == completed["outputs"] + if mode in {"default", "no-wait", "failed"}: + assert captured.out == "" + if mode == "no-wait": + assert "e_test" in captured.err and "running" in captured.err + if mode == "failed": + assert "did not complete successfully" in captured.err else: - assert output == { - "execution_id": "e_test", - "status": "failed" if mode == "failed" else "running", - } + output = json.loads(captured.out) + assert output == (completed if mode == "raw" else completed["outputs"]) assert len(seen) == (1 if mode == "no-wait" else 2) diff --git a/tests/test_transfers.py b/tests/test_transfers.py new file mode 100644 index 0000000..7d5972f --- /dev/null +++ b/tests/test_transfers.py @@ -0,0 +1,112 @@ +"""File transfer contracts, authorization boundaries and failed-download cleanup.""" + +from pathlib import Path +from typing import Iterator + +import httpx +import pytest + +from subfork import APIError, Subfork +from subfork.errors import InvalidResponseError, TransportError + + +def test_upload_returns_artifact_and_streams_multipart(tmp_path: Path) -> None: + """Uploads return server metadata and use the existing upload field.""" + source = tmp_path / "source.txt" + source.write_text("Source document") + + def handle(request: httpx.Request) -> httpx.Response: + """Inspect the upload and the asset-list query.""" + assert request.headers["authorization"] == "Bearer test-key" + if request.method == "GET": + assert request.url.params["include_generated"] == "true" + return httpx.Response(200, json={"assets": []}) + assert request.url.path == "/api/v1/graphs/g_test/assets/files" + assert b'name="upload"; filename="source.txt"' in request.content + assert b"Source document" in request.content + return httpx.Response(201, json={"artifact_id": "art_test", "size_bytes": 15}) + + with Subfork("test-key", transport=httpx.MockTransport(handle)) as client: + assert client.assets.upload("g_test", source)["artifact_id"] == "art_test" + assert client.assets.list("g_test", include_generated=True) == {"assets": []} + + +@pytest.mark.parametrize("redirect", [False, True]) +def test_download_and_overwrite(tmp_path: Path, redirect: bool) -> None: + """Stream bytes and never send the API key or cookies to storage.""" + seen = [] + + def handle(request: httpx.Request) -> httpx.Response: + """Serve a local artifact or a signed storage redirect.""" + seen.append(request) + if request.url.host == "storage.example.com": + assert "authorization" not in request.headers + assert "cookie" not in request.headers + return httpx.Response(200, content=b"MP3 bytes") + assert request.headers["authorization"] == "Bearer test-key" + if redirect: + return httpx.Response( + 302, headers={"location": "https://storage.example.com/audio?signature=test"} + ) + return httpx.Response(200, content=b"MP3 bytes") + + path = tmp_path / "audio.mp3" + with Subfork("test-key", transport=httpx.MockTransport(handle)) as client: + assert client.artifacts.download("art_test", path) == path + assert path.read_bytes() == b"MP3 bytes" + count = len(seen) + with pytest.raises(FileExistsError): + client.artifacts.download("art_test", path) + assert len(seen) == count + client.artifacts.download("art_test", path, overwrite=True) + assert list(tmp_path.iterdir()) == [path] + + +@pytest.mark.parametrize("mode", ["oversize", "http", "insecure", "credentials", "redirect-loop"]) +def test_failed_download_preserves_destination(tmp_path: Path, mode: str) -> None: + """Reject failed, oversized or unsafe transfers without destroying old files.""" + + def handle(request: httpx.Request) -> httpx.Response: + """Return the chosen transfer failure.""" + if mode == "http": + return httpx.Response(409, json={"detail": "private provider data"}) + if mode in {"insecure", "credentials", "redirect-loop"}: + target = { + "insecure": "http://storage.example.com/file", + "credentials": "https://user:pass@storage.example.com/file", + "redirect-loop": "https://storage.example.com/file", + }[mode] + return httpx.Response(302, headers={"location": target}) + return httpx.Response(200, content=b"too many bytes") + + path = tmp_path / "audio.mp3" + path.write_bytes(b"original") + with Subfork("test-key", transport=httpx.MockTransport(handle)) as client: + with pytest.raises((ValueError, APIError, InvalidResponseError)): + client.artifacts.download("art_test", path, overwrite=True, max_bytes=3) + assert path.read_bytes() == b"original" + assert list(tmp_path.iterdir()) == [path] + + +def test_interrupted_download_removes_partial_file(tmp_path: Path) -> None: + """A transport failure cannot replace an existing destination.""" + + class BrokenStream(httpx.SyncByteStream): + """Yield some bytes and then simulate a lost connection.""" + + def __iter__(self) -> Iterator[bytes]: + """Stream a chunk before failing.""" + yield b"partial" * 20000 + raise httpx.ReadError("sensitive-url-must-not-be-reflected") + + def handle(request: httpx.Request) -> httpx.Response: + """Return a stream that fails after writing begins.""" + return httpx.Response(200, stream=BrokenStream()) + + path = tmp_path / "output.mp4" + path.write_bytes(b"original") + with Subfork("test-key", transport=httpx.MockTransport(handle)) as client: + with pytest.raises(TransportError, match="Artifact download could not be completed"): + client.artifacts.download("art_test", path, overwrite=True) + assert path.read_bytes() == b"original" + assert list(tmp_path.iterdir()) == [path]