# 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.

```python
-- 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:

```python
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](#public-buckets).

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`:

```python
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:

```python
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`:

```python
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):

```python
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:

```python
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`:

```python
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:

```python
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.

```python
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.

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

## Removing objects

```python
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:

```python
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:

```python
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:

```python
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](/docs/self-host/configure-infra).

## 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:

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

```bash
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:

```python
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:

```python
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`:

```python
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.

```python
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](/docs/cli/cli-reference).

Under `encore test`, each test gets its own isolated instance of each bucket it uses. See [Testing](/docs/py/develop/testing).