Python SDK Reference

Use the Edgenivo Blob Python SDK to upload, download, inspect, list, copy, and delete blobs from Python applications. The SDK provides an asynchronous API in vercel.blob and a matching synchronous API in vercel.blob.sync.

For JavaScript and TypeScript, see the JS SDK Reference.

Follow Store and serve files from Python with Edgenivo Blob to build private receipt storage with uploads, downloads, and per-user authorization in FastAPI.

Install the Edgenivo Python package. It requires Python 3.10 or later:

uv add vercel

The vercel package includes the Blob SDK from the vercel-blob package. To install only the Blob SDK, add vercel-blob instead. This page documents vercel-blob 0.5.0 and later. Earlier vercel-blob releases on PyPI are an unrelated package that doesn't provide the vercel.blob module.

The examples on this page use the asynchronous API. You can import it as a module:

from vercel import blob

The SDK creates and reuses a default session when you first use it. Most applications don't need to create one. See SDK sessions when you need a custom HTTP client or custom credentials.

By default, the SDK resolves credentials in this order:

1. A Edgenivo OpenID Connect (OIDC) token, from the current Edgenivo request or VERCEL_OIDC_TOKEN, paired with BLOB_STORE_ID
2. BLOB_READ_WRITE_TOKEN, then VERCEL_BLOB_READ_WRITE_TOKEN

If the SDK finds an OIDC token but BLOB_STORE_ID isn't set, and no read-write token is available, it raises BlobCredentialsError. BLOB_STORE_ID can include the store_ prefix.

When you connect a store to a project, Edgenivo adds BLOB_STORE_ID and the OIDC token to the project, so code running on Edgenivo works without configuration. For local development, run vercel link and vercel env pull, then load .env.local into your Python process with a tool such as python-dotenv. When a development OIDC token expires, the SDK refreshes it with your Edgenivo CLI credentials.

To supply credentials from your own configuration, pass a credentials factory to the session. See Configure Blob.

Uploads, copies, and downloads require access="public" or access="private", and the value must match the store's access mode. head_blob(), delete_blob(), query_blobs(), and query_blob_entries() don't take an access argument.

Operations that read, inspect, copy, or delete a blob accept either a pathname or a Blob delivery URL. Uploads and copy destinations accept a pathname.

A pathname is a literal object name relative to the store, not a URL-encoded string or a local file path. The SDK removes one leading slash and rejects:

• Empty pathnames
• Control characters and invalid Unicode
• //
• . and .. path segments

Pathnames for put_blob(), multipart_upload_blob(), and copy_blob() destinations can contain up to 950 UTF-16 code units, the same limit as the JS SDK. Characters outside the Unicode Basic Multilingual Plane count as two units.

The SDK encodes characters such as ?, #, and % when it builds a delivery URL from a pathname. Delivery URLs must use HTTPS and a <store-id>.<access>.blob.vercel-storage.com host. The SDK keeps query strings, removes fragments, and doesn't follow redirects. Custom delivery domains aren't supported.

A private delivery URL must belong to the store identified by your credentials. You can download a public delivery URL without credentials, but a pathname always needs credentials to identify the store.

Upload bytes to a blob:

import asyncio
 
from vercel import blob
 
 
async def main() -> None:
    uploaded = await blob.put_blob(
        "reports/summary.txt",
        b"Quarterly summary",
        access="private",  # or "public"
        add_random_suffix=True,
    )
    print(uploaded.url)
 
 
asyncio.run(main())

put_blob() accepts these arguments:

put_blob() copies in-memory buffers when you call it, so later changes to a bytearray don't affect the upload. It returns a PutResult:

PutResult doesn't include the blob's size or upload time. Use head_blob() to retrieve them. When you set add_random_suffix=True, use the returned pathname or url to refer to the blob.

put_blob() also accepts streaming sources when you pass content_length. Each API accepts only sources of its own shape:

The SDK sends exactly content_length bytes with a Content-Length header and doesn't spool the source to a temporary file. If the source ends early or produces more bytes, the SDK raises BlobContentLengthError and doesn't send the excess data. Uploads with an unknown length aren't supported.

Streaming sources remain yours: the SDK doesn't close, seek, or retry them, and a source's position is unspecified after a failed upload.

The asynchronous API doesn't run blocking reads on worker threads for you, and it raises TypeError for sync readers and sync iterables, such as a file opened with open(). Pass an async byte reader from your async file library, or wrap blocking reads in an async iterable. This example reads a local file on a worker thread:

import asyncio
import os
from collections.abc import AsyncIterator
from typing import BinaryIO
 
from vercel import blob
 
CHUNK_SIZE = 64 * 1024
 
 
async def read_in_thread(file: BinaryIO) -> AsyncIterator[bytes]:
    # Read each chunk on a worker thread so the event loop isn't blocked
    while chunk := await asyncio.to_thread(file.read, CHUNK_SIZE):
        yield chunk
 
 
async def upload_report() -> blob.PutResult:
    with open("report.pdf", "rb") as file:
        return await blob.put_blob(
            "reports/report.pdf",
            read_in_thread(file),
            access="private",
            content_length=os.fstat(file.fileno()).st_size,
        )

Use multipart_upload_blob() when you want to choose each part of a large upload. Enter it with async with, upload parts in order, and the SDK publishes the blob when the block exits successfully:

from collections.abc import AsyncIterator
 
from vercel import blob
 
PART_SIZE = 5 * 1024 * 1024
 
 
async def first_part() -> AsyncIterator[bytes]:
    for _ in range(80):
        yield b"x" * (64 * 1024)
 
 
async def upload_parts() -> blob.PutResult:
    async with blob.multipart_upload_blob(
        "videos/intro.bin",
        access="private",
    ) as upload:
        await upload.upload_part(first_part(), content_length=PART_SIZE)
        await upload.upload_part(b"tail")
 
    return upload.result

multipart_upload_blob() accepts pathname, access, content_type, add_random_suffix, allow_overwrite, and cache_control_max_age with the same meaning as in put_blob(). Don't await the returned upload; enter it as a context manager.

upload_part(body, *, content_length=None) accepts the same bodies as put_blob(). It infers the length of in-memory buffers and requires content_length for streaming sources. The SDK numbers parts in the order you upload them and tracks their entity tags, so there are no part numbers or manual completion step.

Parts follow these rules:

• Each part contains between 1 byte and 5 GiB, and an upload contains at most 10,000 parts.
• Every part except the last one must contain at least 5 MiB. The SDK rejects a part that follows a smaller part before reading its source.
• Upload one part at a time. Overlapping upload_part() calls fail the upload, and the SDK doesn't retry failed parts.
• An upload with no parts raises BlobError on exit. Use put_blob(pathname, b"", ...) to create an empty blob.

A successful exit from the async with block, including an early return or break, publishes the uploaded parts and sets upload.result to a PutResult. An exception or cancellation inside the block prevents publishing. A failed part fails the whole upload, even if you catch its exception, and exiting the block then raises BlobError instead of publishing a partial blob. Reading upload.result before a successful exit raises BlobError.

Download a complete blob into memory:

from vercel import blob
 
result = await blob.get_blob("reports/summary.txt", access="private")
print(result.metadata.content_type)
print(result.body.decode())

get_blob() accepts a pathname or delivery URL and access. It returns a GetResult with metadata, a DownloadMetadata value, and body, the complete content as bytes. The SDK closes the response before returning, and a read error raises an exception instead of returning a partial body.

Read a blob incrementally without buffering it in memory:

from vercel import blob
 
received = 0
async with blob.stream_blob("videos/intro.bin", access="private") as download:
    print(download.metadata.size)
    async for chunk in download:
        received += len(chunk)

stream_blob() accepts the same arguments as get_blob(). Don't await it; enter it with async with. Entering the block sends the request, so download.metadata is available before you read the body.

A download has one consumer. Iterate it once inside its block. Each chunk contains at most 64 KiB. The SDK closes the response at the end of the content, when the block exits early, and on read errors or cancellation. You can also close it with await download.aclose(). The SDK doesn't restart partial downloads.

This FastAPI route streams a private blob to the browser. A background task closes the download after the response finishes:

from contextlib import AsyncExitStack
 
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from starlette.background import BackgroundTask
 
from vercel import blob
 
app = FastAPI()
 
 
@app.get("/api/file")
async def get_file(pathname: str) -> StreamingResponse:
    # Your auth goes here
 
    stack = AsyncExitStack()
    try:
        download = await stack.enter_async_context(
            blob.stream_blob(pathname, access="private")
        )
    except blob.BlobNotFoundError:
        raise HTTPException(status_code=404, detail="Not found") from None
 
    return StreamingResponse(
        download,
        media_type=download.metadata.content_type,
        headers={"X-Content-Type-Options": "nosniff"},
        background=BackgroundTask(stack.aclose),
    )

GetResult.metadata and download.metadata are DownloadMetadata values derived from the response:

Header-derived properties are None when the response omits the header. metadata.size stays None without a Content-Length header, even after get_blob() reads the body, so use len(result.body) for the number of downloaded bytes.

Retrieve a blob's metadata without downloading its content:

from vercel import blob
 
metadata = await blob.head_blob("reports/summary.txt")
print(metadata.size, metadata.uploaded_at)

head_blob() accepts a pathname or delivery URL and returns a HeadResult:

Iterate over the blobs in a store, including blobs in nested folders:

from vercel import blob
 
total_bytes = 0
async for item in blob.query_blobs(prefix="reports/"):
    print(item.pathname, item.size)
    total_bytes += item.size

query_blobs() returns an async iterator and follows pagination cursors automatically. Don't await it; iterate it with async for. It accepts these keyword arguments:

page_size doesn't limit the total number of blobs the iterator yields. To stop early, break out of the loop. The iterator doesn't fetch further pages after you stop.

Creating the iterator doesn't send a request. The SDK validates the arguments and fetches the first page when iteration begins, then fetches each following page only after it yields the current page's blobs. Errors propagate during iteration, and the SDK doesn't retry or restart the listing. A listing isn't a snapshot, so blobs that change while you iterate may or may not appear. Iterate within the session that created the iterator.

Each item is a ListBlob:

ListBlob doesn't include the content type or cache settings. Use head_blob() to retrieve them for a single blob.

Browse one folder level at a time. query_blob_entries() yields ListBlob values for the blobs directly inside the prefix and BlobPrefix values for its subfolders:

from vercel import blob
 
 
async def print_tree(prefix: str = "", indent: str = "") -> None:
    async for entry in blob.query_blob_entries(prefix=prefix):
        match entry:
            case blob.BlobPrefix(prefix=folder):
                print(f"{indent}{folder.removeprefix(prefix)}")
                await print_tree(folder, indent + "  ")
            case blob.ListBlob(pathname=pathname, size=size):
                print(f"{indent}{pathname.removeprefix(prefix)} ({size} bytes)")

query_blob_entries() accepts the same prefix, page_size, and cursor arguments as query_blobs() and paginates the same way. These rules also apply:

• A nonempty prefix must end with /. Omit it, or pass "", to browse the root of the store. The SDK doesn't normalize or URL-decode prefixes.
• BlobPrefix.prefix contains the subfolder's full prefix, including its trailing slash, so you can pass it to another query_blob_entries() call. Folders are virtual groupings of pathnames, not stored blobs, so they have no size or entity tag.
• page_size counts both blobs and folders. Within each page, blobs come before folders, and there's no sort order across pages.

Copy a blob to a new pathname in the same store:

from vercel import blob
 
draft = await blob.head_blob("drafts/getting-started.md")
 
published = await blob.copy_blob(
    draft.url,
    "published/getting-started.md",
    access="private",
    content_type="text/markdown",
    cache_control_max_age=3600,
    if_match=draft.etag,
)
print(published.url)

Edgenivo Blob copies the content on the server, so the SDK doesn't download the source. copy_blob() accepts these arguments:

The copy doesn't inherit the source's content type or cache lifetime. Pass content_type and cache_control_max_age again when the copy needs them.

copy_blob() returns a CopyResult with the same properties as PutResult: url, download_url, pathname, content_type, content_disposition, and etag.

Delete one blob, or up to 1,000 blobs in one request, by pathname or delivery URL:

from vercel import blob
 
await blob.delete_blob("reports/summary.txt")
 
await blob.delete_blob(["reports/2025.pdf", "reports/2026.pdf"])

delete_blob() returns None. Deleting a blob that doesn't exist succeeds, but other failures, such as an unknown store, raise an error. Deleted blobs can remain in the CDN cache for up to one minute.

The SDK validates every target before it sends the request, and it doesn't split larger lists into several requests. Batch deletion isn't atomic, and the SDK doesn't retry it, so a failed request can leave some blobs deleted. Because deleting a missing blob succeeds, you can retry the whole batch.

Pass if_match to delete a blob only if it still has that entity tag. if_match works only with a single target:

from vercel import blob
 
metadata = await blob.head_blob("config.json")
await blob.delete_blob("config.json", if_match=metadata.etag)

The synchronous API mirrors the asynchronous API and doesn't need an event loop. Import it explicitly from vercel.blob:

from vercel.blob import sync as blob
 
uploaded = blob.put_blob("hello.txt", b"Hello, Blob!", access="private")
try:
    result = blob.get_blob(uploaded.url, access="private")
    print(result.body)
 
    with blob.stream_blob(uploaded.url, access="private") as download:
        for chunk in download:
            print(len(chunk))
 
    print(blob.head_blob(uploaded.url).size)
finally:
    blob.delete_blob(uploaded.url)

Remove await and use regular context managers and iterators. Iterate query_blobs() and query_blob_entries() with for, and close a sync download early with download.close().

The sync put_blob() and upload_part() accept sync readers, such as open binary files, and sync iterables:

from pathlib import Path
 
from vercel.blob import sync as blob
 
path = Path("report.pdf")
with path.open("rb") as file:
    uploaded = blob.put_blob(
        "reports/report.pdf",
        file,
        access="private",
        content_length=path.stat().st_size,
    )

Use the asynchronous API unless your application is synchronous. Don't call the synchronous API from an active async event loop.

Catch BlobError to handle any Blob service error. Every BlobError exposes status_code and code when they're available. Catch a more specific type when your application can recover differently:

Invalid arguments, such as a malformed pathname, raise before the SDK resolves credentials or sends a request. The SDK doesn't retry failed requests.

from vercel import blob
 
try:
    metadata = await blob.head_blob("reports/summary.txt")
except blob.BlobNotFoundError:
    metadata = None
except blob.BlobServiceRateLimited as error:
    print(f"Retry in {error.retry_after} seconds")
    raise

The Python SDK supports conditional copies and deletes, but it doesn't currently support conditional uploads or conditional reads, renaming blobs, cache bypass on reads, client uploads, Edgenivo Signed URLs, optimized image uploads, or upload progress callbacks. Use the JS SDK or the Edgenivo CLI for these operations.

An SDK session owns the HTTP connection pool and service clients used by Edgenivo Python SDK calls. The SDK creates and reuses a default session, so most applications should call the SDK directly. Create an explicit session only when you need to customize its HTTP client or Blob's configuration.

Import session from vercel.api and use it as an async or sync context manager. It accepts service_options, a sequence of service-specific options, and httpx_client_factory, a factory for the HTTP client that the session closes when its context exits. See SDK sessions in the Sandbox Python SDK reference for the full parameter reference.

Nested sessions inherit their parent's configuration. Run Blob calls inside their session's scope, including stream_blob() and multipart_upload_blob() blocks, and don't use sync Blob calls inside an async session or async Blob calls inside a sync session.

Pass BlobServiceOptions to configure Blob calls in an async session. This example reads a read-write token and store ID from custom environment variables:

import asyncio
import os
 
from vercel import blob
from vercel.api import session
 
 
async def resolve_credentials() -> blob.BlobCredentials:
    return blob.BlobCredentials(
        token=os.environ["ACME_BLOB_TOKEN"],
        store_id=os.environ["ACME_BLOB_STORE_ID"],
    )
 
 
async def main() -> None:
    options = blob.BlobServiceOptions(credentials_factory=resolve_credentials)
 
    async with session(service_options=[options]):
        metadata = await blob.head_blob("reports/summary.txt")
        print(metadata.size)
 
 
asyncio.run(main())

BlobCredentials accepts token, store_id, and kind. kind defaults to CredentialKind.READ_WRITE, and the store ID must match the store in the read-write token. Set kind=blob.CredentialKind.OIDC to authenticate with an OIDC token. The SDK calls the factory for each operation that needs credentials, so it can return refreshed tokens, but every call within a session must identify the same store.

BlobServiceOptions accepts these parameters:

For sync calls, use SyncBlobServiceOptions from vercel.blob.sync with a synchronous factory and a sync session. The sync API rejects async factories.