Object Storage

Simple and scalable storage for files and unstructured data

Object Storage is a simple and scalable way to store files and unstructured data. The most common implementation is Amazon S3, and its semantics are supported by every major cloud provider.

Encore.py gives you a cloud-agnostic API for object storage. You declare a bucket in code, and Encore provisions it: a local emulator when you run your app with encore run, and Amazon S3 or Google Cloud Storage when you deploy. Every operation is traced automatically.

uploads/storage.py
from encore.storage import objects

profile_pictures = objects.Bucket("profile-pictures")


async def save_picture(user_id: str, data: bytes) -> None:
    await profile_pictures.upload(f"{user_id}.jpeg", data, content_type="image/jpeg")

Creating a bucket

Create a bucket with objects.Bucket, assigned to a module-level variable:

from encore.storage import objects

profile_pictures = objects.Bucket("profile-pictures", versioned=False, public=False)

Encore reads the declaration from your source code to provision the bucket, so a few rules apply:

  • Declare it as a module-level variable, not inside a function, class, or if block.
  • The name must be a string literal. It's a kebab-case name such as profile-pictures, unique within the app.
  • versioned and public are keyword-only and default to False. Write them as True or False literals.

versioned=True keeps previous versions of an object when it is overwritten. public=True makes the bucket's objects readable by anyone at a public URL.

Import the variable from other modules to use the bucket there.

Uploading objects

upload stores bytes under a key, replacing any existing object, and returns the stored object's metadata as an ObjectAttrs:

attrs = await profile_pictures.upload(
    "ada.jpeg",
    data,
    content_type="image/jpeg",
)
print(attrs.size, attrs.etag)

Both options are keyword-only:

  • content_type sets the object's MIME type.
  • if_not_exists=True rejects the upload with PreconditionFailed if an object with that key already exists.

Streaming uploads

To upload a large object without holding it all in memory, pass an async iterable of bytes to upload_stream. It takes the same content_type and if_not_exists options:

from collections.abc import AsyncIterator

from encore.storage import objects

videos = objects.Bucket("videos")


async def read_chunks(path: str) -> AsyncIterator[bytes]:
    with open(path, "rb") as f:
        while chunk := f.read(1024 * 1024):
            yield chunk


async def upload_video(path: str) -> None:
    await videos.upload_stream("intro.mp4", read_chunks(path), content_type="video/mp4")

Chunks are consumed in order, with backpressure: the iterable is only read as fast as the provider accepts data. If the iterable raises or the upload is cancelled, the upload is aborted. Cancellation can't undo an upload the provider has already committed. You own the iterable, so close it yourself if it needs cleanup.

Downloading objects

download returns an object's contents as bytes:

data = await profile_pictures.download("ada.jpeg")

To read part of an object, pass offset (the zero-based start, default 0) and length (the maximum number of bytes, default to the end):

header = await profile_pictures.download("ada.jpeg", offset=0, length=512)

Streaming downloads

download_stream yields the object as chunks of bytes without collecting it in memory. It takes the same version, offset and length options as download. Chunk sizes depend on the provider.

If your loop may stop before the stream is exhausted, wrap it in contextlib.aclosing so the response is released promptly. For example, to copy a large object:

from contextlib import aclosing


async def copy_video(source: str, destination: str) -> None:
    async with aclosing(videos.download_stream(source)) as chunks:
        await videos.upload_stream(destination, chunks, content_type="video/mp4")

Listing objects

list returns an async iterator of ListEntry objects, each with a name, size and etag. Filter by key prefix with prefix, and cap the number of entries with limit:

async for entry in profile_pictures.list(prefix="ada", limit=100):
    print(entry.name, entry.size)

If you stop iterating early, close the iterator with contextlib.aclosing (or call aclose()) to release it promptly:

from contextlib import aclosing


async def first_large_picture() -> str | None:
    async with aclosing(profile_pictures.list()) as entries:
        async for entry in entries:
            if entry.size > 1_000_000:
                return entry.name
    return None

Object attributes

attrs returns an object's metadata: its name, version, size, content_type and etag. It raises ObjectNotFound if the object doesn't exist.

attrs = await profile_pictures.attrs("ada.jpeg")

For convenience, exists returns whether an object exists. It returns False for a missing object and raises on any other storage failure.

if await profile_pictures.exists("ada.jpeg"):
    ...

Removing objects

await profile_pictures.remove("ada.jpeg")

Versioned buckets

In a bucket declared with versioned=True, uploading over an existing object keeps its previous versions. The version of the stored object is in the ObjectAttrs that upload and attrs return.

download, download_stream, attrs, exists and remove take a keyword-only version to work with a specific version instead of the latest one:

documents = objects.Bucket("documents", versioned=True)


async def restore(name: str, version: str) -> None:
    old = await documents.download(name, version=version)
    await documents.upload(name, old)

Public buckets

Declare a bucket with public=True to make its objects publicly readable:

avatars = objects.Bucket("public-avatars", public=True)

public_url returns the URL where an object can be read without authentication. It's synchronous and makes no request:

url = avatars.public_url("ada.jpeg")

Calling public_url on a bucket that isn't declared public is reported as an error when Encore parses your app.

When you deploy with Encore Cloud, public buckets are configured for public access and served through a CDN. When self-hosting, see how to configure public buckets in the infrastructure configuration docs.

Signed URLs

A signed URL grants temporary access to a single object to anyone who has the URL, without any other authentication. It lets clients upload or download content directly to and from the bucket, so large or long-running transfers don't go through your service.

signed_upload_url returns a URL the client can PUT the object's contents to:

url = await profile_pictures.signed_upload_url("ada.jpeg", ttl=7200)

curl -X PUT --data-binary @ada.jpeg "<signed upload URL>"

signed_download_url returns a URL the client can GET the object from, even when the bucket is private:

url = await documents.signed_download_url("letter-1234.pdf", ttl=7200)

ttl is the URL's lifetime in seconds. It defaults to 3600 (one hour) and must be between 1 and 604800 (seven days); anything else raises InvalidArgument.

For objects in a public bucket, prefer public_url: public URLs are served through a CDN, which is usually faster and cheaper.

Using buckets in your code

Encore reads which operations each service performs on a bucket (upload, download, list, and so on) from the method calls in your code. It uses this to draw architecture diagrams and to grant each service exactly the cloud permissions it needs.

For this to work, call the methods on the bucket's module-level variable, directly or through an import:

from uploads.storage import profile_pictures


async def handle_upload(user_id: str, data: bytes) -> None:
    await profile_pictures.upload(f"{user_id}.jpeg", data)

Don't assign the bucket to another variable, return it, or pass it to a function: Encore can't follow it there, and reports the use as an error.

Buckets declared elsewhere

To use a bucket declared by another service, possibly one written in another language, reference it by name with Bucket.named:

from encore.storage import objects

profile_pictures = objects.Bucket.named("profile-pictures")

This doesn't create a bucket. Like a declaration, it must be the value of a module-level variable and the name must be a string literal. Encore reports an error if no bucket with that name is declared in the app.

Error handling

Storage failures raise objects.ObjectsError, a subclass of encore.EncoreError. More specific errors subclass it:

  • objects.ObjectNotFound: the object (or the requested version) doesn't exist.
  • objects.PreconditionFailed: an upload with if_not_exists=True found an existing object.
  • objects.InvalidArgument: an argument is out of range, such as a negative offset or limit, a ttl outside its bounds, or a range that starts beyond the end of the object.
from encore.storage import objects


async def load_picture(user_id: str) -> bytes | None:
    try:
        return await profile_pictures.download(f"{user_id}.jpeg")
    except objects.ObjectNotFound:
        return None

Local development

encore run stores your buckets' objects in a local emulator, so you can develop without any cloud account. Signed URLs and public URLs point at the emulator. The emulator enforces a signed URL's expiry, but doesn't verify its signature.

Inspect and manage the local buckets with the encore objects commands, such as encore objects list-buckets, encore objects list <bucket>, encore objects cat <bucket> <key>, encore objects upload and encore objects rm. See the CLI reference.

Under encore test, each test gets its own isolated instance of each bucket it uses. See Testing.