S3

Import S3Client from rboto. Install the S3 package following Getting started. Configure the client for your bucket's region.

Upload bytes

from rboto import S3Client


async def upload(client: S3Client, bucket: str, key: str) -> None:
    await client.put_object(
        bucket=bucket,
        key=key,
        body=b"hello from rboto",
        content_type="text/plain",
    )

This creates or replaces the object at that key. The body is bytes already held in memory. Multipart upload methods are available as individual service operations.

Download

import asyncio
from pathlib import Path

from rboto import S3Client


async def download(client: S3Client, bucket: str, key: str, destination: Path) -> None:
    response = await client.get_object(bucket=bucket, key=key)
    with destination.open("wb") as output:
        async for chunk in response.body:
            await asyncio.to_thread(output.write, chunk)

A failed download may leave a partial destination file. See Streaming for body ownership, .read(), and memory use.

List every page

Each list call returns one page. Follow its continuation token:

from collections.abc import AsyncIterator

from rboto import S3Client


async def list_keys(client: S3Client, bucket: str) -> AsyncIterator[str]:
    token: str | None = None
    while True:
        response = await client.list_objects_v2(bucket=bucket, continuation_token=token)
        for entry in response.contents or []:
            if entry.key is not None:
                yield entry.key
        token = response.next_continuation_token
        if not token:
            break

Consume this helper with async for. The current Python API does not provide automatic paginator objects.

Catch NoSuchKeyError for a missing-object response. Access policies can cause an access-denied response instead. See Errors.