Upload Handler - Upstash Documentation

An upload handler is the server side of a direct browser upload. It authorizes the upload, signs it, and records what landed. The bytes go from the browser straight to storage and never touch your server, so uploads are not limited by your platform’s request body cap, the 4.5MB one on Vercel or the 6MB one on AWS Lambda.

import "server-only"
import { BlobError, uniquePath, uploadHandler } from "@upstash/blob"

export const uploads = uploadHandler({
  constraints: { maxSize: "20mb", contentTypes: ["image/*", "application/pdf"] },

  onBeforeUpload: async ({ request, file }) => {
    const user = await getUser(request)
    if (!user) throw new BlobError("unauthorized") // 401, nothing is signed
    return { path: uniquePath(`${user.id}/${file.name}`), metadata: { owner: user.id } }
  },

  onUploadComplete: async ({ uploadId, metadata, path, url }) => {
    await db.files.upsert({ uploadId, owner: metadata.owner, path, url })
    return { path }
  },
})

from upstash_blob import AsyncUploadHandler, BeforeUploadResult, BlobError, unique_path


async def authorize(args):
    user = await get_user(args.request)
    if user is None:
        raise BlobError("unauthorized")  # 401, nothing is signed
    return BeforeUploadResult(path=unique_path(f"{user.id}/{args.file.name}"), metadata={"owner": user.id})


async def record(args):
    await db.files.upsert(upload_id=args.upload_id, owner=args.metadata["owner"], path=args.path, url=args.url)
    return {"path": args.path}


uploads = AsyncUploadHandler(
    content_types=["image/*", "application/pdf"],
    max_size="20mb",
    on_before_upload=authorize,
    on_upload_complete=record,
)

uploads, err := blob.NewUploadHandler(blob.UploadHandlerOptions[struct{}]{
	ContentTypes: []string{"image/*", "application/pdf"},
	MaxSize:      20 * blob.MB,

	OnBeforeUpload: func(
		a blob.BeforeUploadArgs[struct{}],
	) (blob.BeforeUploadResult, error) {
		user := getUser(a.Request)
		if user == nil {
			// 401, nothing is signed
			return blob.BeforeUploadResult{}, blob.NewError(
				blob.CodeUnauthorized,
				blob.ErrorOptions{},
			)
		}
		return blob.BeforeUploadResult{
			Path:     blob.UniquePath(user.ID + "/" + a.File.Name),
			Metadata: map[string]string{"owner": user.ID},
		}, nil
	},

	OnUploadComplete: func(a blob.UploadCompleteArgs[struct{}]) (any, error) {
		err := db.UpsertFile(a.Request.Context(), File{
			UploadID: a.UploadID,
			Owner:    a.Metadata["owner"],
			Path:     a.Path,
			URL:      a.URL,
		})
		if err != nil {
			return nil, err
		}
		return map[string]string{"path": a.Path}, nil
	},
})

The handler runs a before-upload callback at the begin phase, before anything is signed, and a completion callback at the end phase, once the object exists. The PUTs in between go to storage, not to you. The phases are listed on Types. Option names, argument types and framework wiring: TypeScript, Python, Go.

Mounting it

// app/api/upload/route.ts (Next.js App Router)
import { uploads } from "@/lib/uploads"

export const { GET, POST } = uploads

# app/main.py (FastAPI)
from fastapi import FastAPI
from upstash_blob.fastapi import upload_view

from app.uploads import uploads

app = FastAPI()
app.add_api_route("/api/upload", upload_view(uploads), methods=["GET", "POST"])

// main.go (net/http)
http.Handle("/api/upload", uploads)

The handler answers GET and POST at one URL. POST runs the upload phases and GET serves the route’s constraints. The browser client assumes /api/upload and can be pointed anywhere else. It speaks the same protocol to every server SDK, so the frontend does not change with the backend language. Mounting in other frameworks: TypeScript, Python, Go. If the bytes have to pass through your app instead, write an ordinary route that calls put and drive it with useServerUpload.

Handler options

Option	Default	Description
Bucket	one bucket built from `UPSTASH_BLOB_TOKEN`	The bucket every route writes to. Pass one when the token lives under another variable, the bucket needs a default `cache`, or the platform has no environment variables.
Constraints	none, anything	What the route accepts. Served by `GET` and enforced at `begin`. See Constraints.
Multipart	the multipart threshold	Where an upload stops being one PUT and starts going up in parts.
Endpoint	none	Where the handler is mounted. Only needed to separate two handlers on one bucket.
Context	none	Runs once per POST. Its value reaches every callback. See Context.
Input	none	A schema that validates what the browser sends as `input` before the before-upload callback runs. See Input and state.
Before-upload callback	required, here or on every route	Authorizes the upload and names the path.
Completion callback	none	Records the object. What it returns becomes `upload.blob.data` in the browser.
Error callback	none	Sees every refusal. The one place to log.
Routes	none, one unnamed route	Mounts several routes at this one endpoint.

Everything except routes, endpoint and context is a default that each route inherits. A route overrides the keys it names, so a handler with five routes states the shared policy once. Constraints merge one level deeper; see Per-route constraints. The before-upload callback is the only required one. Bad options fail at startup, where they are written: an unparseable multipart size, an invalid route name, an empty routes map, a missing token.

Authorizing an upload

onBeforeUpload: async ({ ctx, route, request, file, input }) => {
  return { path: uniquePath(`${ctx.userId}/${file.name}`), metadata: { owner: ctx.userId } }
}

def authorize(args: BeforeUploadArgs) -> BeforeUploadResult:
    # args.ctx, args.route, args.request, args.file, args.input
    return BeforeUploadResult(
        path=unique_path(f"{args.ctx.user_id}/{args.file.name}"),
        metadata={"owner": args.ctx.user_id},
    )

func authorize(
	a blob.BeforeUploadArgs[*Session],
) (blob.BeforeUploadResult, error) {
	// a.Ctx, a.Route, a.Request, a.File, a.Input
	return blob.BeforeUploadResult{
		Path:     blob.UniquePath(a.Ctx.UserID + "/" + a.File.Name),
		Metadata: map[string]string{"owner": a.Ctx.UserID},
	}, nil
}

The before-upload callback runs at begin, before anything is signed and before any bytes exist. It decides whether the upload happens and where the object goes. It receives the context value, the route name, the begin request with its headers and cookies, the file the browser declared (name, type and size, before a byte was sent), and the validated input when the route declares a schema. It returns:

Field	Default	Description
Path	required	Where the object is stored.
Cache	the bucket default	The `Cache-Control` this object is stored with. See Caching.
Metadata	none	Signed into the upload and handed back to the completion callback.
Constraints	the route’s	Narrows this one upload’s limits.
State	none	Anything computed here that the completion and error callbacks should get without a lookup. See Input and state.

The declared file type is what the object is stored and served as. Whether the bytes match is checked separately; see Byte sniffing.

Paths

uniquePath("chat/Thread_42/Holiday Pic.PNG")
// chat/Thread_42/Holiday Pic-7Kd2mQ9x.PNG

unique_path("chat/Thread_42/Holiday Pic.PNG")
# chat/Thread_42/Holiday Pic-7Kd2mQ9x.PNG

blob.UniquePath("chat/Thread_42/Holiday Pic.PNG")
// chat/Thread_42/Holiday Pic-7Kd2mQ9x.PNG

The path is required, and may not contain . or .. segments. The unique path helper preserves IDs and filenames, including case, underscores, spaces, and Unicode, and adds one random suffix to the final filename. Slashes anywhere in the string are directories, so a prefix can be included as is. The rules are on Unique paths. A stable path is an overwrite. Two concurrent uploads to the same path race, and the loser’s end can fail with not_found. Use a stable path only when overwriting is the intent.

Metadata

Metadata is signed into the presigned PUT, so the browser cannot add to it or change it. It comes back on the completion callback, is stored on the object, and is readable later with info. Same rules as a server-side write: lowercase keys, printable ASCII values. See Metadata. The key upstash-upload is reserved for the SDK and is refused with invalid_input.

Narrowing per user

Return constraints to tighten the route’s limits for this one upload, for example a smaller max size on a free plan. They can only make the route stricter. See Narrowing per user.

Refusing

onBeforeUpload: async ({ request, file }) => {
  const user = await getUser(request)
  if (!user) throw new BlobError("unauthorized")
  if (await overQuota(user)) throw new BlobError("too_large", { message: "you are out of space" })
  return { path: uniquePath(`${user.id}/${file.name}`) }
}

async def authorize(args: BeforeUploadArgs) -> BeforeUploadResult:
    user = await get_user(args.request)
    if user is None:
        raise BlobError("unauthorized")
    if await over_quota(user):
        raise BlobError("too_large", "you are out of space")
    return BeforeUploadResult(path=unique_path(f"{user.id}/{args.file.name}"))

func authorize(
	a blob.BeforeUploadArgs[struct{}],
) (blob.BeforeUploadResult, error) {
	user := getUser(a.Request)
	if user == nil {
		return blob.BeforeUploadResult{}, blob.NewError(
			blob.CodeUnauthorized,
			blob.ErrorOptions{},
		)
	}
	if overQuota(user) {
		return blob.BeforeUploadResult{}, blob.NewError(
			blob.CodeTooLarge,
			blob.ErrorOptions{Message: "you are out of space"},
		)
	}
	path := blob.UniquePath(user.ID + "/" + a.File.Name)
	return blob.BeforeUploadResult{Path: path}, nil
}

Refuse with a BlobError (a returned *blob.Error in Go). Nothing is signed and no URL is handed out. The error reaches the browser with its code intact, so a hook can switch on error.code instead of status numbers. The codes are on Errors. The browser never retries begin, so a callback that writes a row runs once per file.

Recording the upload

onUploadComplete: async ({ uploadId, path, url, size, contentType, metadata, state, ctx }) => {
  await db.files.upsert({ uploadId, owner: ctx.userId, path, url, size, contentType })
  return { path }
}

async def record(args: UploadCompleteArgs) -> dict[str, str]:
    await db.files.upsert(
        upload_id=args.upload_id,
        owner=args.ctx.user_id,
        path=args.path,
        url=args.url,
        size=args.size,
        content_type=args.content_type,
    )
    return {"path": args.path}

func record(a blob.UploadCompleteArgs[*Session]) (any, error) {
	err := db.UpsertFile(a.Request.Context(), File{
		UploadID:    a.UploadID,
		Owner:       a.Ctx.UserID,
		Path:        a.Path,
		URL:         a.URL,
		Size:        a.Size,
		ContentType: a.ContentType,
	})
	if err != nil {
		return nil, err
	}
	return map[string]string{"path": a.Path}, nil
}

The completion callback runs at end, once the object exists. It receives the completed object’s fields plus everything this route knew about the upload:

Field	Description
Path	Where the object is stored.
URL	The public URL. Not set on a private bucket.
Versioned URL	The URL with the etag on the query. For a stable path that gets overwritten.
Size	Bytes actually stored, verified against what the browser declared.
Etag	The stored object’s etag.
Uploaded at	When storage wrote it.
Content type	What the object is stored as. This is the one to record.
Context	What context returned for this request.
Route	The route name.
Request	The `end` request.
File	What the browser declared at `begin`. The original filename survives only here.
Upload id	Identifies this upload. Stable across retries: the idempotency key.
Multipart upload id	For aborting this exact multipart upload later; see Aborting one upload. Not set for a single PUT.
Metadata	What the before-upload callback returned, minus the SDK’s marker.
State	What the before-upload callback returned as state.

What it returns is handed to the browser as upload.blob.data.

Retries and throws

It may run more than once. The browser retries end on a network failure, so upsert on the upload id, which is stable across retries, rather than inserting a new row.A throw deletes the object, and so does a returned error in Go. That is the intent for a refusal, and a trap for a database error: a short outage destroys bytes that uploaded fine. Catch your own storage errors rather than letting them escape.

onUploadComplete: async ({ uploadId, path, url, metadata }) => {
  try {
    await db.files.upsert({ uploadId, owner: metadata.owner, path, url })
  } catch (e) {
    console.error("[uploads] could not record", path, e)
    // Throw only when losing the file is better than keeping an unrecorded one.
    throw new BlobError("not_ready", { message: "could not record the upload, try again" })
  }
  return { path }
}

async def record(args: UploadCompleteArgs) -> dict[str, str]:
    try:
        await db.files.upsert(upload_id=args.upload_id, owner=args.metadata["owner"], path=args.path, url=args.url)
    except Exception:
        logger.exception("[uploads] could not record %s", args.path)
        # Raise only when losing the file is better than keeping an unrecorded one.
        raise BlobError("not_ready", "could not record the upload, try again")
    return {"path": args.path}

func record(a blob.UploadCompleteArgs[struct{}]) (any, error) {
	err := db.UpsertFile(a.Request.Context(), File{
		UploadID: a.UploadID,
		Owner:    a.Metadata["owner"],
		Path:     a.Path,
		URL:      a.URL,
	})
	if err != nil {
		slog.Error("[uploads] could not record", "path", a.Path, "err", err)
		// Return an error only when losing the file is better than
		// keeping an unrecorded one.
		return nil, blob.NewError(blob.CodeNotReady, blob.ErrorOptions{
			Message: "could not record the upload, try again",
		})
	}
	return map[string]string{"path": a.Path}, nil
}

A browser that dies before end leaves an object no callback recorded. See Abandoned uploads.

Handling errors

onError: ({ ctx, route, request, error, file, path, metadata, state }) => {
  logger.warn({ route, path, user: ctx?.userId, error })
  if (error instanceof PaymentRequired) return new BlobError("forbidden", { message: "plan expired" })
}

def on_error(args: UploadErrorArgs) -> BlobError | None:
    logger.warning("upload refused: route=%s path=%s error=%r", args.route, args.path, args.error)
    if isinstance(args.error, PaymentRequired):
        return BlobError("forbidden", "plan expired")
    return None

func onError(a blob.UploadErrorArgs[*Session]) error {
	slog.Warn("upload refused", "route", a.Route, "path", a.Path, "err", a.Err)
	var expired PaymentRequired
	if errors.As(a.Err, &expired) {
		return blob.NewError(
			blob.CodeForbidden,
			blob.ErrorOptions{Message: "plan expired"},
		)
	}
	return nil
}

The error callback sees every refusal this endpoint produces, including the handler’s own and a request for a route nobody mounted. Log here and keep the other callbacks about the happy path. It receives the context value (not set when context itself failed, or when no route matched), the route name from the query even when nothing mounts it, the request, the error, and as much of the file, path, metadata and state as the request had reached before it failed. Return a BlobError or a response to answer with it. Return nothing and the answer is left alone. Written on the handler it is the default for every route; a route with its own replaces it. How a thrown error becomes a response is on Errors.

Context

export const uploads = uploadHandler({
  context: (request) => requireUser(request), // throws BlobError('unauthorized') on a dead session

  routes: {
    avatar: {
      onBeforeUpload: ({ ctx, file }) => ({ path: uniquePath(`avatars/${ctx.id}/${file.name}`) }),
      onUploadComplete: ({ ctx, url }) => db.users.update(ctx.id, { avatarUrl: url }),
    },
    attachment: {
      onBeforeUpload: ({ ctx, file }) => ({ path: uniquePath(`files/${ctx.id}/${file.name}`) }),
    },
  },
})

async def save_avatar(args: UploadCompleteArgs) -> None:
    await db.users.update(args.ctx.id, avatar_url=args.url)


uploads = AsyncUploadHandler(
    context=require_user,  # raises BlobError("unauthorized") on a dead session
    routes={
        "avatar": UploadRoute(
            on_before_upload=lambda args: BeforeUploadResult(
                path=unique_path(f"avatars/{args.ctx.id}/{args.file.name}"),
            ),
            on_upload_complete=save_avatar,
        ),
        "attachment": UploadRoute(
            on_before_upload=lambda args: BeforeUploadResult(
                path=unique_path(f"files/{args.ctx.id}/{args.file.name}"),
            ),
        ),
    },
)

uploads, err := blob.NewUploadHandler(blob.UploadHandlerOptions[*User]{
	// returns an unauthorized *blob.Error on a dead session
	Context: requireUser,

	Routes: map[string]blob.UploadRoute[*User]{
		"avatar": {
			OnBeforeUpload: func(
				a blob.BeforeUploadArgs[*User],
			) (blob.BeforeUploadResult, error) {
				name := a.Ctx.ID + "/" + a.File.Name
				path := blob.UniquePath("avatars/" + name)
				return blob.BeforeUploadResult{Path: path}, nil
			},
			OnUploadComplete: func(
				a blob.UploadCompleteArgs[*User],
			) (any, error) {
				return nil, db.SetAvatar(a.Request.Context(), a.Ctx.ID, a.URL)
			},
		},
		"attachment": {
			OnBeforeUpload: func(
				a blob.BeforeUploadArgs[*User],
			) (blob.BeforeUploadResult, error) {
				path := blob.UniquePath("files/" + a.Ctx.ID + "/" + a.File.Name)
				return blob.BeforeUploadResult{Path: path}, nil
			},
		},
	},
})

Context runs once per POST, after the route is picked and before any body is read. Its value reaches every callback. It does not run for GET, which serves a public constraints document. Use it when several routes share one auth check, or when the completion and error callbacks need an authenticated value. With a single route, authorizing inside the before-upload callback and carrying an id in metadata is shorter.

Multiple routes

export const uploads = uploadHandler({
  constraints: { maxSize: "20mb", contentTypes: ["image/*"] },
  context: (request) => requireUser(request),

  routes: {
    avatar: {
      constraints: { maxSize: "2mb" },
      onBeforeUpload: ({ ctx }) => ({ path: `avatars/${ctx.id}`, cache: "revalidate" }),
    },
    attachment: {
      constraints: { contentTypes: null }, // clears the handler's list for this route
      multipart: "50mb",
      onBeforeUpload: ({ ctx, file }) => ({ path: uniquePath(`files/${ctx.id}/${file.name}`) }),
    },
  },
})

uploads = AsyncUploadHandler(
    max_size="20mb",
    content_types=["image/*"],
    context=require_user,
    routes={
        "avatar": UploadRoute(
            max_size="2mb",
            on_before_upload=lambda args: BeforeUploadResult(path=f"avatars/{args.ctx.id}", cache="revalidate"),
        ),
        "attachment": UploadRoute(
            content_types=None,  # clears the handler's list for this route
            multipart="50mb",
            on_before_upload=lambda args: BeforeUploadResult(
                path=unique_path(f"files/{args.ctx.id}/{args.file.name}"),
            ),
        ),
    },
)

uploads, err := blob.NewUploadHandler(blob.UploadHandlerOptions[*User]{
	MaxSize:      20 * blob.MB,
	ContentTypes: []string{"image/*"},
	Context:      requireUser,

	Routes: map[string]blob.UploadRoute[*User]{
		"avatar": {
			MaxSize: 2 * blob.MB,
			OnBeforeUpload: func(
				a blob.BeforeUploadArgs[*User],
			) (blob.BeforeUploadResult, error) {
				return blob.BeforeUploadResult{
					Path:  "avatars/" + a.Ctx.ID,
					Cache: blob.CacheRevalidate,
				}, nil
			},
		},
		"attachment": {
			// clears the handler's list for this route
			ContentTypes: []string{},
			Multipart:    blob.MultipartOver(50 * blob.MB),
			OnBeforeUpload: func(
				a blob.BeforeUploadArgs[*User],
			) (blob.BeforeUploadResult, error) {
				path := blob.UniquePath("files/" + a.Ctx.ID + "/" + a.File.Name)
				return blob.BeforeUploadResult{Path: path}, nil
			},
		},
	},
})

app/page.tsx

const avatar = useUpload("avatar")
const attachment = useUpload("attachment")

Routes mount several routes at one endpoint. The name travels in the query as ?route=avatar, and the client passes the name instead of a URL.

Route names must match /^[A-Za-z_][\w-]*$/, checked when the handler is built.
An unknown name is a 404 that does not list the routes the handler mounts. It still reaches the error callback.
An upload authorized by one route cannot be completed at another.
A handler with no routes is itself the route. It is reached with no ?route=.
Two handlers on the same bucket that mount the same route names need an endpoint to tell them apart.

Input and state

const thread = uploadRoute<Session>()({
  input: z.object({ threadId: z.string().uuid() }),

  onBeforeUpload: ({ ctx, input, file }) => ({
    path: uniquePath(`chat/${input.threadId}/${file.name}`),
    state: { threadId: input.threadId, name: file.name },
  }),

  onUploadComplete: ({ ctx, state, url, uploadId }) =>
    db.messages.insert({ uploadId, threadId: state.threadId, name: state.name, owner: ctx.id, url }),
})

class Thread(BaseModel):
    threadId: UUID


async def post_message(args: UploadCompleteArgs) -> None:
    await db.messages.insert(
        upload_id=args.upload_id,
        thread_id=args.state["thread_id"],
        name=args.state["name"],
        owner=args.ctx.id,
        url=args.url,
    )


thread = UploadRoute(
    input=Thread,
    on_before_upload=lambda args: BeforeUploadResult(
        path=unique_path(f"chat/{args.input.threadId}/{args.file.name}"),
        state={"thread_id": str(args.input.threadId), "name": args.file.name},
    ),
    on_upload_complete=post_message,
)

type Thread struct {
	ThreadID string `json:"threadId"`
}

type threadState struct {
	ThreadID string `json:"threadId"`
	Name     string `json:"name"`
}

thread := blob.UploadRoute[*User]{
	Input: blob.JSONInput[Thread](),

	OnBeforeUpload: func(
		a blob.BeforeUploadArgs[*User],
	) (blob.BeforeUploadResult, error) {
		t := a.Input.(Thread)
		if _, err := uuid.Parse(t.ThreadID); err != nil {
			return blob.BeforeUploadResult{}, blob.NewError(
				blob.CodeInvalidInput,
				blob.ErrorOptions{Message: "the threadId is not a UUID"},
			)
		}
		return blob.BeforeUploadResult{
			Path:  blob.UniquePath("chat/" + t.ThreadID + "/" + a.File.Name),
			State: threadState{ThreadID: t.ThreadID, Name: a.File.Name},
		}, nil
	},

	OnUploadComplete: func(a blob.UploadCompleteArgs[*User]) (any, error) {
		var s threadState
		if err := json.Unmarshal(a.State, &s); err != nil {
			return nil, err
		}
		ctx := a.Request.Context()
		return nil, db.InsertMessage(
			ctx, a.UploadID, s.ThreadID, s.Name, a.Ctx.ID, a.URL,
		)
	},
}

app/page.tsx

const { start } = useUpload("thread")
start({ file, input: { threadId } })

A route can take input, data the browser sends along with the file, and pass state from the before-upload callback to the completion callback.

input is validated before the before-upload callback runs, and only the parsed value reaches it. A route with no schema refuses any input the browser sends with invalid_input. Validation failures are invalid_input too, with the issues joined as path: message, so a bad threadId reads threadId: Invalid uuid.
state is for values the callback already computed and does not want to look up again. It travels through the browser and is readable in devtools, so put a row id there, never a secret. See The completion token.

How a route declares its schema and types its state: TypeScript, Python, Go.

The GET endpoint

GET on the route serves its constraints as JSON. That is what fills accept and constraints on the hook, and it lets the hook refuse an oversized file before any request leaves the browser. The server still enforces the same limits at begin. The document and its caching are on Constraints.

​Mounting it

​Handler options

​Authorizing an upload

​Paths

​Metadata

​Narrowing per user

​Refusing

​Recording the upload

​Retries and throws

​Handling errors

​Context

​Multiple routes

​Input and state

​The GET endpoint

Mounting it

Handler options

Authorizing an upload

Paths

Metadata

Narrowing per user

Refusing

Recording the upload

Retries and throws

Handling errors

Context

Multiple routes

Input and state

The GET endpoint