DX-3128: upload handler, protocol version, and blob-js parity for release

.github/workflows/release.yml

@@ -10,6 +10,9 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - uses: astral-sh/setup-uv@v6
+      - run: uv sync --locked
+      - run: uv run ruff check . && uv run ruff format --check . && uv run mypy upstash_blob
+      - run: uv run pytest tests/unit
       - run: uv build
       - run: uv publish --token "${{ secrets.PYPI_TOKEN }}"
       - name: Create GitHub release

README.md

@@ -57,7 +57,10 @@ async with AsyncBucket.from_env() as bucket:
 - `delete(prefix="")` needs `all=True`. A partial list delete raises `partial_delete`.
 - Metadata is printable ASCII; anything else is refused with `invalid_input`.
 - `cache` takes `"immutable"`, `"revalidate"`, `"no-store"`, a duration (`"15m"`, `3600`,
-  `timedelta`), or a verbatim header. Sizes are decimal (`"2mb"` is 2,000,000 bytes).
+  `timedelta`), or a verbatim header.
+- Sizes count bytes. Decimal units (`KB`, `MB`, `GB`, `TB`) are powers of 1,000 and binary units
+  (`KiB`, `MiB`, `GiB`, `TiB`) powers of 1,024, case-insensitive: `"32MB"` is 32,000,000 bytes and
+  refuses a 32 MiB file, which needs `"32MiB"` or `32 * 1024 * 1024`.
 
 ### Signed URLs
 
@@ -71,6 +74,49 @@ Links are signed by Upstash and live at most 10 minutes (read default: 5); `expi
 per link. `headers` on an upload URL are pinned into the signature. Whether a bucket is private is
 decided in the console: a private bucket has no `url` on any blob and `public_url()` is None.
 
+### Browser uploads
+
+`UploadHandler` (sync) and `AsyncUploadHandler` serve the protocol of `@upstash/blob/browser` and
+`@upstash/blob/react`, so a React app can upload straight to storage through a Python backend, with
+multipart, resume, progress and a completion callback.
+
+```python
+from upstash_blob import AsyncUploadHandler, BeforeUploadResult, UploadRoute, unique_path
+from upstash_blob.fastapi import upload_view
+
+uploads = AsyncUploadHandler(
+    context=require_user,  # its return is `ctx`; raise a 401 to refuse
+    on_before_upload=lambda a: BeforeUploadResult(path=unique_path(f"u/{a.ctx.id}/{a.file.name}")),
+    on_upload_complete=lambda a: {"url": a.url},  # reaches the browser as upload.blob.data
+    routes={
+        "avatar": UploadRoute(content_types=["image/*"], max_size="5mb"),
+        "video": UploadRoute(content_types=["video/*"], max_size="2gb"),
+    },
+)
+app.add_api_route("/api/upload", upload_view(uploads), methods=["GET", "POST"])
+```
+
+```tsx
+const upload = useUpload('avatar');  // POSTs to /api/upload?route=avatar
+```
+
+- Django: `path("api/upload", upload_view(uploads))` from `upstash_blob.django`, with either
+  handler. Flask: `app.add_url_rule("/api/upload", view_func=upload_view(uploads), methods=["GET", "POST"])`
+  from `upstash_blob.flask`, with `UploadHandler`. Anything else: `handler.handle(method, request=...,
+  route=..., body=..., headers=..., native=<framework response class>)` returns an `UploadResponse`.
+- Omit `bucket=` to use `UPSTASH_BLOB_TOKEN`. Every handler option is a default a route may replace;
+  `content_types=None` or `max_size=None` on a route clears the handler's.
+- `input=` validates what the browser sends as `input`: a pydantic model, a `TypeAdapter`, or a
+  callable that raises `ValueError`.
+- `on_upload_complete` runs at least once per upload; `upload_id` is stable across deliveries.
+  Raising deletes the object. `state` rides in a signed but readable token: an id, never a secret.
+- `on_error` sees every refusal; return a `BlobError` or a framework response to answer with it.
+- `unique_path` adds an eight-character random suffix to the final filename and keeps everything else
+  as given: `u/7/Q3 Report.pdf` becomes `u/7/Q3 Report-<random>.pdf`. A path with a `.` or `..`
+  segment is refused as `invalid_input`. Store the returned path and authorize reads by it.
+- The handler speaks the same versioned upload protocol as `@upstash/blob`'s `uploadHandler`, so
+  any release of the browser client works against it and either side can be upgraded first.
+
 ### Incomplete uploads
 
 A multipart upload that was never finished is invisible to `list()` and blocks bucket deletion. The

pyproject.toml

@@ -35,6 +35,9 @@ Issues = "https://github.com/upstash/blob-py/issues"
 [dependency-groups]
 dev = [
     "boto3>=1.35",
+    "django>=4.2",
+    "fastapi>=0.110",
+    "flask>=3.0",
     "mypy>=1.14",
     "pytest>=8.3",
     "pytest-asyncio>=0.24",
@@ -65,6 +68,10 @@ per-file-ignores = { "tests/**" = ["E501"] }
 strict = true
 python_version = "3.10"
 
+[[tool.mypy.overrides]]
+module = ["django.*"]
+ignore_missing_imports = true
+
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
 testpaths = ["tests"]

tests/fake.py

@@ -123,6 +123,15 @@ def r2(self, req: httpx.Request) -> httpx.Response:
             return xml(
                 f"<CompleteMultipartUploadResult><ETag>{esc(self.objects[key].etag)}</ETag></CompleteMultipartUploadResult>"
             )
+        elif m == "GET" and "uploadId" in q:
+            up = self.uploads.get(q["uploadId"])
+            if not up:
+                return error(404, "NoSuchUpload")
+            items = "".join(
+                f"<Part><PartNumber>{n}</PartNumber><ETag>{esc(etag_of(d))}</ETag><Size>{len(d)}</Size></Part>"
+                for n, d in sorted(up[2].items())
+            )
+            return xml(f"<ListPartsResult><IsTruncated>false</IsTruncated>{items}</ListPartsResult>")
         elif m == "DELETE" and "uploadId" in q:
             return httpx.Response(204) if self.uploads.pop(q["uploadId"], None) else error(404, "NoSuchUpload")
         elif m == "PUT" and "x-amz-copy-source" in req.headers:

tests/live/test_upload_live.py

@@ -0,0 +1,138 @@
+import json
+from typing import Any, Dict, List
+
+import httpx
+import pytest
+
+from tests.live.conftest import p
+from upstash_blob import (
+    AsyncBucket,
+    AsyncUploadHandler,
+    BeforeUploadResult,
+    BlobError,
+    Bucket,
+    UploadCompleteArgs,
+    UploadHandler,
+)
+
+pytestmark = pytest.mark.live
+
+MIB = 1024 * 1024
+JSON = {"content-type": "application/json"}
+
+
+def data(n: int, seed: int = 7) -> bytes:
+    return bytes((i * 31 + seed) & 0xFF for i in range(n))
+
+
+def post(h: UploadHandler, body: Dict[str, Any]) -> Any:
+    res = h.handle("POST", body=json.dumps(body).encode(), headers=JSON)
+    assert res.status == 200, res.body
+    return json.loads(res.body)
+
+
+def browser_put(begun: Dict[str, Any], body: bytes) -> List[Dict[str, Any]]:
+    """What @upstash/blob/browser does with the plan: PUT each part to its url, verbatim headers."""
+    size = begun["upload"]["partSize"]
+    landed = []
+    for part in begun["upload"]["parts"]:
+        n = part["n"]
+        chunk = body[(n - 1) * size : n * size]
+        res = httpx.put(part["url"], content=chunk, headers=part.get("headers", {}), timeout=60)
+        assert res.status_code == 200, res.text
+        landed.append({"n": n, "etag": res.headers["etag"]})
+    return landed
+
+
+def test_single_put_through_the_handler(pub: Bucket) -> None:
+    seen: List[UploadCompleteArgs] = []
+    h = UploadHandler(
+        bucket=pub,
+        content_types=["text/plain"],
+        on_before_upload=lambda a: BeforeUploadResult(path=p(f"h/{a.file.name}"), metadata={"by": "u1"}),
+        on_upload_complete=lambda a: seen.append(a) or {"ok": True},
+    )
+    body = b"hello from the handler"
+    begun = post(h, {"phase": "begin", "file": {"name": "a.txt", "type": "text/plain", "size": len(body)}})
+    assert begun["upload"]["multipart"] is False
+    landed = browser_put(begun, body)
+    done = post(h, {"phase": "end", "completionToken": begun["completionToken"], "parts": landed})
+    assert done["data"] == {"ok": True}
+    assert done["blob"]["size"] == len(body) and done["blob"]["contentType"] == "text/plain"
+    assert httpx.get(done["blob"]["url"]).content == body
+    assert seen[0].metadata == {"by": "u1"}
+    got = pub.get(p("h/a.txt"))
+    assert got.content == body and got.metadata["by"] == "u1"
+
+
+def test_a_tampered_put_is_refused_by_storage(pub: Bucket) -> None:
+    h = UploadHandler(bucket=pub, on_before_upload=lambda a: BeforeUploadResult(path=p("h/tamper.txt")))
+    begun = post(h, {"phase": "begin", "file": {"name": "t.txt", "type": "text/plain", "size": 5}})
+    part = begun["upload"]["parts"][0]
+    headers = {**part["headers"], "x-amz-meta-upstash-upload": "someone-else"}
+    assert httpx.put(part["url"], content=b"hello", headers=headers).status_code == 403
+    assert httpx.put(part["url"], content=b"hello!", headers=part["headers"]).status_code == 403
+
+
+def test_refusal_at_end_deletes_the_object(priv: Bucket) -> None:
+    def refuse(a: UploadCompleteArgs) -> None:
+        raise BlobError("forbidden", "no")
+
+    h = UploadHandler(
+        bucket=priv, on_before_upload=lambda a: BeforeUploadResult(path=p("h/refused.txt")), on_upload_complete=refuse
+    )
+    begun = post(h, {"phase": "begin", "file": {"name": "r.txt", "type": "text/plain", "size": 3}})
+    assert begun["upload"]["parts"][0]["headers"]["cache-control"].startswith("private")
+    landed = browser_put(begun, b"abc")
+    assert priv.exists(p("h/refused.txt"))
+    body = {"phase": "end", "completionToken": begun["completionToken"], "parts": landed}
+    res = h.handle("POST", body=json.dumps(body).encode(), headers=JSON)
+    assert res.status == 403
+    assert not priv.exists(p("h/refused.txt"))
+
+
+def test_multipart_through_the_handler_with_resume(pub: Bucket) -> None:
+    size = 6 * MIB + 123
+    body = data(size)
+    h = UploadHandler(
+        bucket=pub,
+        multipart=True,
+        on_before_upload=lambda a: BeforeUploadResult(path=p("h/big.bin"), metadata={"k": "v"}),
+    )
+    begun = post(h, {"phase": "begin", "file": {"name": "big.bin", "type": "application/octet-stream", "size": size}})
+    plan = begun["upload"]
+    assert plan["multipart"] is True and plan["partSize"] == 5 * MIB and len(plan["parts"]) == 2
+    first = browser_put({**begun, "upload": {**plan, "parts": plan["parts"][:1]}}, body)
+    resumed = post(h, {"phase": "parts", "completionToken": begun["completionToken"], "from": 2})
+    assert resumed["landed"] == first
+    rest = browser_put({**begun, "upload": {**plan, "parts": resumed["parts"]}}, body)
+    done = post(h, {"phase": "end", "completionToken": begun["completionToken"], "parts": first + rest})
+    assert done["blob"]["size"] == size
+    got = pub.get(p("h/big.bin"))
+    assert got.content == body and got.metadata["k"] == "v"
+
+
+def test_multipart_cancel_aborts(pub: Bucket) -> None:
+    h = UploadHandler(bucket=pub, multipart=True, on_before_upload=lambda a: BeforeUploadResult(path=p("h/cancel.bin")))
+    begun = post(h, {"phase": "begin", "file": {"name": "c", "type": "", "size": 6 * MIB}})
+    assert any(u.path == p("h/cancel.bin") for u in pub.list_multipart_uploads(prefix=p("h/")))
+    post(h, {"phase": "cancel", "completionToken": begun["completionToken"]})
+    assert not any(u.path == p("h/cancel.bin") for u in pub.list_multipart_uploads(prefix=p("h/")))
+
+
+async def test_async_handler_single_put() -> None:
+    import os
+
+    if not os.environ.get("UPSTASH_BLOB_PUBLIC_TOKEN"):
+        pytest.skip("UPSTASH_BLOB_PUBLIC_TOKEN is not set")
+    async with AsyncBucket.from_env("UPSTASH_BLOB_PUBLIC_TOKEN") as bucket:
+        h = AsyncUploadHandler(bucket=bucket, on_before_upload=lambda a: BeforeUploadResult(path=p("h/async.txt")))
+        body = b"async hello"
+        file = {"name": "x.txt", "type": "text/plain", "size": len(body)}
+        res = await h.handle("POST", body=json.dumps({"phase": "begin", "file": file}).encode(), headers=JSON)
+        begun = json.loads(res.body)
+        landed = browser_put(begun, body)
+        end = {"phase": "end", "completionToken": begun["completionToken"], "parts": landed}
+        res = await h.handle("POST", body=json.dumps(end).encode(), headers=JSON)
+        assert res.status == 200, res.body
+        assert (await bucket.get(p("h/async.txt"))).content == body