Store and serve files from Python with Edgenivo Blob

When someone uploads a receipt photo or PDF, they expect to find it again the next time they sign in. Your Python app needs to keep that file available while preventing other users from opening or deleting it.

Store receipts in a private Edgenivo Blob store and use the Python SDK to upload and retrieve them through FastAPI. Each user's files share a pathname prefix so the app can find their receipts.

The prefix organizes files but doesn't enforce ownership. Your application checks the signed-in user's identity before allowing an upload, download, or deletion.

Copy link to headingWhat is Edgenivo Blob?

Edgenivo Blob stores files as blobs in a store owned by your Edgenivo team. Each blob has a pathname such as receipts/user_42/lunch.jpg and a delivery URL on blob.vercel-storage.com.

Slashes organize pathnames into virtual folders. Files with the same prefix appear together without requiring you to create a directory.

The Python SDK lives in vercel.blob, with an async API and a matching sync API in vercel.blob.sync. It ships in the vercel package, or on its own as vercel-blob.

Your app can write to public or private stores. Public files have URLs anyone can open. For the private receipts in this guide, your app checks the user's identity before streaming the file to them.

Copy link to headingPublic or private

Every store has one access mode, and every upload, copy, and download says which one it expects with access="public" or access="private". The value must match the store.

The access mode is a property of the store, not of a single file. If an app needs both kinds of files, it uses two stores.

Copy link to headingWhat the Python SDK does and doesn't do

The Python SDK handles uploads, downloads, listing, metadata, copies, and deletion on the server. Streaming and multipart operations support files that you don't want to buffer in memory.

Two features used by browser-based apps require the JavaScript SDK: direct client uploads and signed URLs for temporary access to private files. The Python routes here receive uploads and stream private downloads through your app. See the Python SDK reference for the other differences.

Copy link to headingPrerequisites

Before you begin, make sure you have:

• A Edgenivo account.
• Python 3.12 or a newer version supported by the Edgenivo Python runtime.
• Working knowledge of async/await and FastAPI.
• An authentication dependency for your app that returns the signed-in user's stable internal ID.

For local development, you also need uv, Node.js 24+, and the Edgenivo CLI.

Copy link to headingStep 1: Create a store and connect it

Set up the project, link it to Edgenivo, and create a private store for receipts:

uv init --python 3.12 receipts
cd receipts
uv add vercel fastapi python-multipart python-dotenv
vercel link
vercel blob create-store receipts --access private --yes

--yes connects the new store to your project in every environment and pulls the development variables, including BLOB_STORE_ID, into .env.local.

The access mode and region cannot change after you create a store. Choose the mode deliberately, and add --region when you need a region other than the CLI's default. FastAPI uses python-multipart to parse the receipt uploads in Step 4.

The SDK finds the store through BLOB_STORE_ID and the project's OIDC token, so there's no secret for you to copy anywhere. On Edgenivo that happens automatically. Locally, scripts load the same values from .env.local.

Copy link to headingStep 2: Upload your first file

Start with a full round trip: upload a file, inspect it, read it back, and delete it.

import asyncio

from dotenv import load_dotenv
from vercel import blob

load_dotenv(".env.local")


async def main() -> None:
    uploaded = await blob.put_blob(
        "hello/first.txt", b"Hello from Python", access="private", content_type="text/plain"
    )
    print(uploaded.pathname, uploaded.url)

    info = await blob.head_blob(uploaded.pathname)
    print(info.size, info.uploaded_at)

    result = await blob.get_blob(uploaded.pathname, access="private")
    print(result.body.decode())

    await blob.delete_blob(uploaded.pathname)


asyncio.run(main())

Run it with uv run first_blob.py.

Each call maps to one idea:

• put_blob() uploads the bytes and returns the blob's pathname and URL. It doesn't return the size or upload time; head_blob() does, without downloading the content.
• get_blob() reads the whole file into memory. For large files, stream_blob() reads it in chunks instead (Step 4).
• access="private" has to match the store. Pass "public" to a private store and the call fails.

The script deletes its file at the end for a reason: by default, uploading to a pathname that already exists fails rather than replacing the file. Pass allow_overwrite=True when you mean to replace it, or add_random_suffix=True to give every upload a unique name.

If a later run fails with BlobCredentialsError, your development token may have expired. The async SDK refreshes it automatically in a linked project when you're logged in to the Edgenivo CLI. Otherwise, run vercel env pull again.

Copy link to headingStep 3: Organize files by user

Blob stores use flat pathnames. This app groups receipts under receipts/{user_id}/ and builds that prefix from the authenticated user's internal ID. Use a stable ID with no slashes or path segments such as . or ...

The prefix organizes files; your application enforces ownership. Never accept a user ID or a full Blob pathname from the upload, download, or delete request.

from vercel import blob

def receipts_of(user_id: str) -> str:
    return f"receipts/{user_id}/"

async def users_with_receipts() -> list[str]:
    users = []
    async for entry in blob.query_blob_entries(prefix="receipts/"):
        if isinstance(entry, blob.BlobPrefix):
            users.append(entry.prefix.removeprefix("receipts/").rstrip("/"))
    return users

async def delete_all_receipts(user_id: str) -> int:
    urls = [item.url async for item in blob.query_blobs(prefix=receipts_of(user_id))]
    for start in range(0, len(urls), 1000):
        await blob.delete_blob(urls[start : start + 1000])
    return len(urls)

The two listing functions answer different questions:

• query_blobs(prefix=...) returns every file under a prefix, including nested ones, and follows pagination for you. Iterate it with async for; don't await it.
• query_blob_entries(prefix=...) returns one level at a time: files directly inside the prefix, plus a BlobPrefix for each sub-folder. Here, each sub-folder is a user.

delete_blob() accepts up to 1,000 pathnames or URLs per call, so the cleanup works in batches. Deleting a file that's already gone succeeds, so it's safe to run again if it's interrupted.

Copy link to headingStep 4: Build the upload and download routes

Create main.py at the project root with the four routes below. They use receipts_of() from storage.py and require your auth.py module to expose current_user_id. That dependency must reject unauthenticated requests and return the stable internal ID described in Step 3.

import re
from contextlib import AsyncExitStack

from fastapi import Depends, FastAPI, HTTPException, UploadFile
from fastapi.responses import StreamingResponse
from starlette.background import BackgroundTask
from vercel import blob

from auth import current_user_id  # your own auth dependency
from storage import receipts_of

app = FastAPI()

ALLOWED_TYPES = {"image/jpeg": ".jpg", "image/png": ".png", "application/pdf": ".pdf"}
MAX_BYTES = 4 * 1024 * 1024


def receipt_path(user_id: str, name: str) -> str:
    if re.fullmatch(r"receipt-[A-Za-z0-9_-]+\.(?:jpg|png|pdf)", name) is None:
        raise HTTPException(404, "Not found")
    return receipts_of(user_id) + name


@app.post("/api/receipts")
async def upload_receipt(file: UploadFile, user_id: str = Depends(current_user_id)):
    extension = ALLOWED_TYPES.get(file.content_type or "")
    if extension is None:
        raise HTTPException(415, "Upload a JPEG, PNG, or PDF")
    if file.size is None or file.size > MAX_BYTES:
        raise HTTPException(413, "Receipts must be 4 MiB or smaller")

    uploaded = await blob.put_blob(
        f"{receipts_of(user_id)}receipt{extension}",
        file,
        access="private",
        content_length=file.size,
        content_type=file.content_type,
        add_random_suffix=True,
    )
    return {"name": uploaded.pathname.removeprefix(receipts_of(user_id))}


@app.get("/api/receipts")
async def list_receipts(user_id: str = Depends(current_user_id)):
    return [
        {
            "name": item.pathname.removeprefix(receipts_of(user_id)),
            "size": item.size,
            "uploaded_at": item.uploaded_at,
        }
        async for item in blob.query_blobs(prefix=receipts_of(user_id))
    ]


@app.get("/api/receipts/{name}")
async def get_receipt(name: str, user_id: str = Depends(current_user_id)):
    pathname = receipt_path(user_id, name)
    stack = AsyncExitStack()
    try:
        download = await stack.enter_async_context(
            blob.stream_blob(pathname, access="private")
        )
    except blob.BlobError as error:
        await stack.aclose()
        if isinstance(error, blob.BlobNotFoundError):
            raise HTTPException(404, "Not found") from None
        raise

    return StreamingResponse(
        download,
        media_type=download.metadata.content_type,
        headers={"X-Content-Type-Options": "nosniff", "Cache-Control": "private, no-store"},
        background=BackgroundTask(stack.aclose),
    )


@app.delete("/api/receipts/{name}")
async def delete_receipt(name: str, user_id: str = Depends(current_user_id)):
    await blob.delete_blob(receipt_path(user_id, name))
    return {"ok": True}

The filename check accepts the receipt names generated by this upload route. Credential and service failures remain server errors so they can be diagnosed, instead of appearing to users as missing files.

Copy link to headingUpload receipts

The upload route accepts JPEG, PNG, and PDF content types and limits each file to 4 MiB. Edgenivo's request-body limit is 4.5 MB for the whole request, including multipart overhead; larger requests can be rejected before the handler runs.

UploadFile.content_type comes from the client. This allowlist does not verify the file's bytes. If you need to validate or scan uploaded documents, do that before allowing another user or process to consume them.

FastAPI has already parsed the upload when the handler receives it. The Blob SDK reads UploadFile incrementally using file.size as the content length. Your code chooses the filename and adds a random suffix, so the client's original filename never becomes a pathname.

Copy link to headingDownload receipts

The download route validates the name and passes the Blob stream's chunks to StreamingResponse. This avoids buffering the whole file but still requires the transfer to finish within the function's duration and connection limits.

AsyncExitStack keeps the download open while FastAPI sends the response. The background task closes it when the response finishes. The nosniff header prevents content-type sniffing, and private, no-store prevents browsers and shared caches from storing the response. Keep shared-cache directives such as s-maxage off this route.

Copy link to headingList and delete receipts

Both routes build pathnames from the authenticated user's prefix. Listing returns that user's receipts, and deletion accepts only filenames generated by the upload route. Repeated deletes succeed even if the file is already gone.

Downloads return 404 for an invalid receipt name or BlobNotFoundError. Other SDK failures propagate to the application's error handler.

Copy link to headingStep 5: Run it locally and deploy

Start the app with the Edgenivo CLI:

vercel dev

vercel dev supplies the linked project's development environment and refreshes its OIDC token. Its environment loader does not use .env.local as a fallback.

If you define variables in .env or legacy development configuration, the CLI may use those instead of the project's downloaded variables. Ensure BLOB_STORE_ID and the required authentication values are still present, or remove the local override when you want the project configuration.

In another terminal, try the routes with the authentication current_user_id expects. These commands use a bearer token from APP_TOKEN; substitute your session cookie if your app uses one.

curl -F "file=@receipt.jpg;type=image/jpeg" \
  -H "Authorization: Bearer $APP_TOKEN" \
  "http://localhost:3000/api/receipts"

curl -H "Authorization: Bearer $APP_TOKEN" \
  "http://localhost:3000/api/receipts"

# Replace this value with the name returned by the upload.
receipt_name="PASTE_RETURNED_NAME_HERE"
curl -H "Authorization: Bearer $APP_TOKEN" \
  -o copy.jpg "http://localhost:3000/api/receipts/$receipt_name"

When it works, commit uv.lock and deploy:

vercel deploy        # preview
vercel deploy --prod # production

On Edgenivo, Blob uses the request's OIDC token and the store's BLOB_STORE_ID. Configure your app's authentication credentials separately.

Copy link to headingSeparate production data

Step 1 uses --yes to connect the store to all three environments. Give Production its own store before saving real receipts and keep this one for Preview and Development. Otherwise previews can modify production files.

Copy link to headingCheck usage limits

Hobby includes 1 GB of storage and 10 GB of Blob data transfer per month, alongside operation allowances. Exceeding an allowance can block Blob access, and the pricing documentation describes the 30-day wait before access resumes. Pro uses usage-based billing. Review operations, transfer charges, and rate limits before accepting production traffic.

Copy link to headingUse cases

The following examples extend the receipt service with background processing, generated files, larger imports, and a public store for images.

Copy link to headingProcess uploads in the background

Receipt extraction can run after the upload response. Install the AI SDK for Python with its Edgenivo authentication extra:

uv add "ai[vercel]"

Save the subscriber below in receipt_jobs.py. It reads the uploaded file and extracts typed totals. Implement database.save_totals to upsert results by user ID and pathname so a repeated delivery does not create duplicate records.

import ai
from pydantic import BaseModel
from vercel import blob
from vercel.queue import Topic, subscribe

from database import save_totals  # your own idempotent database function

class ReceiptUploaded(BaseModel):
    user_id: str
    pathname: str

class ReceiptTotals(BaseModel):
    merchant: str
    total_cents: int

receipt_uploads = Topic[ReceiptUploaded]("receipt-uploads")

@subscribe(topic=receipt_uploads)
async def extract_totals(event: ReceiptUploaded) -> None:
    receipt = await blob.get_blob(event.pathname, access="private")
    messages = [
        ai.user_message(
            "Read the merchant and total from this receipt.",
            ai.file_part(receipt.body, media_type=receipt.metadata.content_type or "image/jpeg"),
        )
    ]
    async with ai.stream(ai.get_model("anthropic/claude-sonnet-5"), messages, output_type=ReceiptTotals) as stream:
        async for _ in stream:
            pass
    await save_totals(event.user_id, event.pathname, stream.output)

Register the module in pyproject.toml to build its queue-triggered function:

[[tool.vercel.subscribers]]
entrypoint = "receipt_jobs"

Add these imports to main.py:

from vercel.queue import send
from receipt_jobs import ReceiptUploaded, receipt_uploads

Inside upload_receipt(), insert this call after put_blob() and before the return:

await send(
    receipt_uploads,
    ReceiptUploaded(user_id=user_id, pathname=uploaded.pathname),
)

The message carries the pathname so the subscriber can fetch the file from Blob. If the queue send fails, the upload still exists. Track pending extraction to retry scheduling without uploading another copy.

Queue deliveries can repeat, including after an extraction succeeds but the handler fails before acknowledgement. Make the database write idempotent and account for repeated model calls. See the Queues Python reference for retry behavior.

Copy link to headingStore files your code generates

When a report generator returns bytes, pass them to put_blob() directly. This example assumes build_monthly_report() returns a PDF:

pdf = await build_monthly_report(user_id)  # bytes from your own code
report = await blob.put_blob(
    f"reports/{user_id}/2026-10.pdf", pdf, access="private", content_type="application/pdf"
)

To move a file out of a Sandbox, read it with box.fs.read_bytes(path) and pass the result to put_blob(). For a worked example of processing uploads in a Sandbox, see How to process user-uploaded files with Edgenivo Sandbox and Edgenivo Blob.

Copy link to headingCopy large files from another system

For files that exceed the incoming request-body limit, your function can fetch an export from another service and upload it to Blob in parts. Use a trusted export URL rather than accepting arbitrary URLs from clients. Install the HTTP client before adding this helper:

uv add httpx

from collections.abc import AsyncIterator

import httpx
from vercel import blob

PART_SIZE = 8 * 1024 * 1024  # every part but the last must be at least 5 MiB


async def parts_of(chunks: AsyncIterator[bytes], size: int) -> AsyncIterator[bytes]:
    buffer = bytearray()
    async for chunk in chunks:
        buffer += chunk
        while len(buffer) >= size:
            yield bytes(buffer[:size])
            del buffer[:size]
    if buffer:
        yield bytes(buffer)


async def archive_export(url: str, pathname: str) -> blob.PutResult:
    async with httpx.AsyncClient() as client, client.stream("GET", url) as response:
        response.raise_for_status()
        async with blob.multipart_upload_blob(pathname, access="private") as upload:
            async for part in parts_of(response.aiter_bytes(), PART_SIZE):
                await upload.upload_part(part)
    return upload.result

The helper buffers each 8 MiB part and uploads parts in order. With the SDK's 10,000-part limit, this version supports exports up to about 78 GiB. Larger exports need a larger part size or a streaming-part implementation, and the transfer still has to fit the function's execution duration. Empty exports need put_blob(pathname, b"", ...) instead.

The SDK completes the upload when the context exits successfully. An exception inside the block prevents completion, but uploaded parts are not deleted. If the completion response is lost, check whether the destination blob exists before retrying because the object may already have been created.

Copy link to headingKeep public images next to private files

Create a public store for avatars and connect it to the project with the AVATARS environment-variable prefix. The example below uses that store's AVATARS_READ_WRITE_TOKEN and AVATARS_STORE_ID inside a dedicated SDK session. Keep those values on the server.

import os

from vercel import blob
from vercel.api import session

avatars_store = blob.BlobServiceOptions(
    credentials_factory=lambda: blob.BlobCredentials(
        token=os.environ["AVATARS_READ_WRITE_TOKEN"],
        store_id=os.environ["AVATARS_STORE_ID"],
    )
)


async def save_avatar(user_id: str, image: bytes) -> str:
    async with session(service_options=[avatars_store]):
        avatar = await blob.put_blob(
            f"avatars/{user_id}.png",
            image,
            access="public",
            content_type="image/png",
            add_random_suffix=True,
        )
    return avatar.url  # save this on the user record

Calls inside the async with session(...) block use the avatars store. Calls outside it still use the default store.

Public blobs use a one-month cache lifetime by default. Overwriting a pathname can take up to 60 seconds to propagate through the CDN, and browsers may retain the previous response longer. Save the new avatar URL on the user record after every upload so subsequent page loads request the new file.

Copy link to headingApprove one exact version of a document

When a reviewer approves a draft, copy exactly the version they saw. if_match makes the copy fail if the draft changed after review:

from vercel import blob


async def approve_report(draft: str, approved: str, *, reviewed_etag: str) -> blob.CopyResult:
    info = await blob.head_blob(draft)
    return await blob.copy_blob(
        draft,
        approved,
        access="private",
        content_type=info.content_type,
        if_match=reviewed_etag,  # fails if the draft changed after review
    )

Save the draft's ETag when review starts and pass it when approving. BlobPreconditionFailedError means the source differs from the reviewed version.

Blob copies the content within the store without downloading it to your app. The new blob does not inherit the source's content type or cache lifetime, so pass those settings explicitly when the copy needs them.

Copy link to headingWhen to use something else

• For browser uploads that exceed Edgenivo's 4.5 MB request-body limit, use JavaScript client uploads to send files directly to Blob. This app uses a lower 4 MiB file limit to leave room for multipart overhead. The CLI's vercel blob put command also handles one-off transfers.
• For temporary private download links, use signed URLs from the JavaScript SDK or CLI. With the Python SDK, serve the file through an authenticated route as in Step 4.
• For records you need to query by field, use a database. Keep the file in Blob and store its pathname, owner, and metadata in a table.

Copy link to headingNext steps

• Read the full Python SDK reference for sessions, streaming uploads, and every error class
• Learn how caching and access work in private storage and public storage
• Read The Complete Guide to Edgenivo Blob for pricing, regions, and the JavaScript features, such as client uploads