Python

Use pushduck from FastAPI, Django, Flask, Starlette or Litestar — with the same JavaScript client

The Python package is a server. Your frontend keeps using the JavaScript client unchanged — it only speaks the wire protocol, so it neither knows nor cares which language answered.

pip install pushduck
# or
uv add pushduck

Zero runtime dependencies. Signing is HMAC-SHA256 from hashlib and the protocol is JSON over HTTP, so nothing is dragged into your application.

Defining routes

A route is a value. Every field is a point where your application participates, and every one is optional.

upload.py

import os
from pushduck import Router, Route, UploadConfig, image, file

config = UploadConfig(
    bucket=os.environ["AWS_S3_BUCKET"],
    region=os.environ["AWS_REGION"],
    access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
    secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
    session_token=os.environ.get("AWS_SESSION_TOKEN"),  # ECS, EKS IRSA, OIDC
)

router = Router(config)

router.add("imageUpload", Route(
    schema=image(max_size="5MB"),
    authorize=[require_session],                       # raise to reject
    user=load_user,                                    # becomes ctx.user
    storage_path=lambda ctx, f: f"{ctx.user.tenant}/{f.name}",  # a fragment, not the whole path
    metadata=lambda ctx, f: {"user_id": ctx.user.id},  # the upload's metadata
    on_complete=[record_upload],
))

router.add("documentUpload", Route(
    schema=file(max_size="50MB", allow_types=["application/pdf"]),
    metadata=lambda ctx, f: {"tenant": current_tenant()},   # sync is fine too
))

The channels

Channel	Runs	What it does
`authorize`	once per request	Raise to reject the request. The return value is ignored.
`user`	once per request	Its return value is `ctx.user`.
`around`	once per request	`async def f(ctx): ... yield ...` — wraps everything below.
`validate`	per file	Raise to fail that file; the rest of the batch continues.
`storage_path`	per file	Returns a path fragment. pushduck sanitises it.
`metadata`	per file	Its return value is the upload's metadata.
`on_complete`	per file	After an upload is confirmed.
`on_error`	on failure	Observational.

Channels that produce a value take one callable. Channels that veto, wrap or observe take a list, applied in order. Sync and async are both accepted everywhere, resolved once when the route is registered.

Each value-producing channel is named after the value it produces. That is why "authenticate, but publish no metadata" has an obvious spelling — a route with authorize and no metadata — rather than being something you have to express by returning nothing.

ctx.metadata starts empty, and the client's payload is never merged into it. What the caller sent is available as ctx.client_metadata, under a name that reads as a warning. A route with no metadata channel publishes {}. To forward what the client sent, say so: metadata=lambda ctx, f: dict(ctx.client_metadata).

Keys are sanitised, always

The storage_path channel returns a fragment. pushduck rejects .. segments, absolute paths and URL delimiters, then re-sanitises every segment before signing:

storage_path=lambda ctx, f: f"{ctx.user.tenant}/{f.name}"
# "acme/写真 photo.png"  ->  "acme/写真_photo.png"
# "../../etc/passwd"    ->  CONFIG_INVALID, refused

This is deliberate and not adjustable. A path hook that is trusted verbatim is how CVE-2024-39330 (Django) and CVE-2026-34750 (Payload CMS) happened; both were fixed by moving the check somewhere an override cannot reach.

A route is a plain dataclass, so dataclasses.replace derives one from another and stays correct when a channel is added later:

from dataclasses import replace

tenant = Route(authorize=[require_session], user=load_user, storage_path=tenant_key)

router.add("avatar",   replace(tenant, schema=image(max_size="5MB")))
router.add("document", replace(tenant, schema=file(max_size="50MB")))

For behaviour every route shares, give the router defaults. They prepend to each route's own channels rather than replacing them:

router = Router(config, defaults=Route(authorize=[require_session]))

Seeing what is registered

print(router.describe())
#   avatar    authorize(1) user storage_path metadata on_complete(1)
#   public    schema only

Worth printing at boot. A route that has quietly lost its authentication is otherwise invisible until someone notices the uploads.

Raise UploadError to control the status. Any other exception becomes a 500 with its message withheld, because a driver exception routinely carries hostnames, credentials and query fragments that nobody vetted for disclosure.

Mounting

main.py

from fastapi import FastAPI
from upload import router

app = FastAPI()
app.mount("/api/upload", router.asgi())

main.py

from starlette.applications import Starlette
from starlette.routing import Mount
from upload import router

app = Starlette(routes=[Mount("/api/upload", router.asgi())])

Django's URLconf expects a view, not an ASGI app — so path("api/upload", router.asgi()) does not work. The adapter is a few lines:

uploads/views.py

import asyncio
from django.http import HttpResponse
from pushduck import Request
from .upload import router

def upload_view(request):
    response = asyncio.run(
        router.handle(
            Request(
                method=request.method,
                path=request.get_full_path(),
                query=request.GET.dict(),
                headers={k.lower(): v for k, v in request.headers.items()},
                body=request.body,
            )
        )
    )

    django_response = HttpResponse(
        response.body,
        status=response.status,
        content_type=response.headers.get("Content-Type", "application/json"),
    )
    for key, value in response.headers.items():
        if key.lower() != "content-type":
            django_response[key] = value
    return django_response

urls.py

from django.urls import path
from uploads.views import upload_view

urlpatterns = [path("api/upload", upload_view)]

Flask is WSGI, so it takes the other adapter. Either keep the endpoint out of Flask's routing entirely:

app.py

from flask import Flask
from werkzeug.middleware.dispatcher import DispatcherMiddleware
from upload import router

app = Flask(__name__)
app.wsgi_app = DispatcherMiddleware(app.wsgi_app, {"/api/upload": router.wsgi()})

…or write an ordinary view, which most people will find more familiar:

app.py

import asyncio
from flask import Flask, request as flask_request
from pushduck import Request
from upload import router

app = Flask(__name__)

@app.route("/api/upload", methods=["GET", "POST"])
def upload():
    response = asyncio.run(
        router.handle(
            Request(
                method=flask_request.method,
                path=flask_request.full_path,
                query=flask_request.args.to_dict(),
                headers={k.lower(): v for k, v in flask_request.headers.items()},
                body=flask_request.get_data(),
            )
        )
    )
    return response.body, response.status, list(response.headers.items())

Every snippet above is executed by tests/test_frameworks.py against the real framework, with a request driven through it and the response asserted. A snippet nobody runs is a claim, not documentation — and this project has shipped incorrect ones before.

Any other ASGI framework — Litestar, Quart, Sanic — mounts router.asgi() the same way FastAPI does, and any other WSGI framework takes router.wsgi(). Those are not listed above only because they are not covered by a test here.

The frontend is unchanged

app/upload.ts

import { createUploadClient } from "pushduck/client";

const upload = createUploadClient({ endpoint: "/api/upload" });

Multipart, resume, progress and typed errors all work exactly as they do against the TypeScript server, because the client only speaks the protocol.

What is supported


Presign, complete, introspection	yes
Multipart uploads	yes — init, sign, complete, abort, list
Resume after interruption	yes, verified against real storage
Completion tokens	yes
Temporary credentials (STS)	yes
S3-compatible providers	yes — R2, MinIO, Spaces, path-style addressing
Resume across server restarts	no server-side session store yet

Behaviour is pinned by the shared conformance suite, which this server passes in full — the same 23 cases the TypeScript and Go servers pass.

Python

On this page