S3 Storage¶

The S3Storage backend stores files in Amazon S3 or any S3-compatible object storage service. It supports presigned URLs, server-side operations, and streaming for efficient handling of large files.

Supported Services¶

S3Storage works with:

Amazon S3 - The original S3 service
Cloudflare R2 - S3-compatible with zero egress fees
DigitalOcean Spaces - Simple S3-compatible storage
MinIO - Self-hosted S3-compatible storage
Backblaze B2 - S3-compatible with B2-native features
Linode Object Storage - S3-compatible from Akamai
Wasabi - Hot cloud storage with S3 compatibility

Installation¶

Install with the S3 extra:

pip install litestar-storages[s3]

This installs aioboto3 for async S3 operations.

Configuration¶

S3Config Options¶

from datetime import timedelta
from litestar_storages import S3Storage, S3Config

config = S3Config(
    bucket="my-bucket",                      # Required: bucket name
    region="us-east-1",                      # Optional: AWS region
    endpoint_url=None,                       # Optional: custom endpoint for S3-compatible
    access_key_id=None,                      # Optional: falls back to env/IAM
    secret_access_key=None,                  # Optional: falls back to env/IAM
    session_token=None,                      # Optional: for temporary credentials
    prefix="",                               # Optional: key prefix for all operations
    presigned_expiry=timedelta(hours=1),     # Optional: default URL expiration
    use_ssl=True,                            # Optional: use HTTPS
    verify_ssl=True,                         # Optional: verify SSL certificates
    max_pool_connections=10,                 # Optional: connection pool size
)

storage = S3Storage(config)

Configuration Options¶

Option	Type	Default	Description
`bucket`	`str`	Required	S3 bucket name
`region`	`str \| None`	`None`	AWS region (e.g., “us-east-1”)
`endpoint_url`	`str \| None`	`None`	Custom endpoint URL for S3-compatible services
`access_key_id`	`str \| None`	`None`	AWS access key ID (falls back to environment)
`secret_access_key`	`str \| None`	`None`	AWS secret access key (falls back to environment)
`session_token`	`str \| None`	`None`	AWS session token for temporary credentials
`prefix`	`str`	`""`	Key prefix applied to all operations
`presigned_expiry`	`timedelta`	1 hour	Default expiration for presigned URLs
`use_ssl`	`bool`	`True`	Use HTTPS for connections
`verify_ssl`	`bool`	`True`	Verify SSL certificates
`max_pool_connections`	`int`	`10`	Maximum connections in pool

AWS S3 Setup¶

Basic Configuration¶

from litestar_storages import S3Storage, S3Config

# Using explicit credentials
storage = S3Storage(
    config=S3Config(
        bucket="my-app-uploads",
        region="us-west-2",
        access_key_id="AKIAIOSFODNN7EXAMPLE",
        secret_access_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
    )
)

# Using environment variables (recommended)
# Set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY in environment
storage = S3Storage(
    config=S3Config(
        bucket="my-app-uploads",
        region="us-west-2",
    )
)

IAM Role (EC2, ECS, Lambda)¶

When running on AWS infrastructure, use IAM roles instead of explicit credentials:

# No credentials needed - uses IAM role attached to the instance/container
storage = S3Storage(
    config=S3Config(
        bucket="my-app-uploads",
        region="us-west-2",
    )
)

Required IAM permissions:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:PutObject",
                "s3:GetObject",
                "s3:DeleteObject",
                "s3:ListBucket",
                "s3:GetObjectAttributes"
            ],
            "Resource": [
                "arn:aws:s3:::my-app-uploads",
                "arn:aws:s3:::my-app-uploads/*"
            ]
        }
    ]
}

S3-Compatible Services¶

Cloudflare R2¶

storage = S3Storage(
    config=S3Config(
        bucket="my-bucket",
        endpoint_url="https://<account-id>.r2.cloudflarestorage.com",
        access_key_id="your-r2-access-key-id",
        secret_access_key="your-r2-secret-access-key",
        region="auto",  # R2 uses "auto" for region
    )
)

Find your R2 credentials in the Cloudflare dashboard under R2 > Manage R2 API Tokens.

DigitalOcean Spaces¶

storage = S3Storage(
    config=S3Config(
        bucket="my-space",
        endpoint_url="https://nyc3.digitaloceanspaces.com",
        access_key_id="your-spaces-key",
        secret_access_key="your-spaces-secret",
        region="nyc3",
    )
)

Available regions: nyc3, sfo3, ams3, sgp1, fra1

MinIO (Self-Hosted)¶

storage = S3Storage(
    config=S3Config(
        bucket="my-bucket",
        endpoint_url="http://minio.local:9000",
        access_key_id="minioadmin",
        secret_access_key="minioadmin",
        region="us-east-1",  # Can be any value for MinIO
        use_ssl=False,       # Disable for local development
        verify_ssl=False,
    )
)

Backblaze B2¶

storage = S3Storage(
    config=S3Config(
        bucket="my-bucket",
        endpoint_url="https://s3.us-west-004.backblazeb2.com",
        access_key_id="your-application-key-id",
        secret_access_key="your-application-key",
        region="us-west-004",
    )
)

Presigned URLs¶

Presigned URLs provide time-limited access to private files without exposing credentials.

Generating Presigned URLs¶

from datetime import timedelta

# Use default expiration (from config)
url = await storage.url("documents/report.pdf")

# Custom expiration
url = await storage.url(
    "documents/report.pdf",
    expires_in=timedelta(minutes=15),
)

# Long-lived URL (use sparingly)
url = await storage.url(
    "public/image.jpg",
    expires_in=timedelta(days=7),
)

Presigned URL Patterns¶

Download links in API responses:

from litestar import get
from litestar_storages import Storage


@get("/files/{key:path}/download-url")
async def get_download_url(
    key: str,
    storage: Storage,
) -> dict[str, str]:
    """Generate a temporary download URL."""
    url = await storage.url(key, expires_in=timedelta(minutes=30))
    return {"download_url": url, "expires_in": "30 minutes"}

Direct browser downloads:

from litestar import get
from litestar.response import Redirect


@get("/files/{key:path}/download")
async def download_file(
    key: str,
    storage: Storage,
) -> Redirect:
    """Redirect to presigned URL for download."""
    url = await storage.url(key, expires_in=timedelta(minutes=5))
    return Redirect(url)

Key Prefixes¶

Use prefixes to organize files within a single bucket:

# All operations will be prefixed with "app-name/uploads/"
storage = S3Storage(
    config=S3Config(
        bucket="shared-bucket",
        prefix="app-name/uploads/",
    )
)

# Stores at: s3://shared-bucket/app-name/uploads/images/photo.jpg
await storage.put("images/photo.jpg", data)

# Lists only files under the prefix
async for file in storage.list("images/"):
    print(file.key)  # Returns "images/photo.jpg", not full path

This is useful for:

Multiple applications sharing a bucket
Environment separation (production/, staging/)
Tenant isolation in multi-tenant applications

Usage Examples¶

File Upload with Metadata¶

from litestar import post
from litestar.datastructures import UploadFile
from litestar_storages import Storage, StoredFile


@post("/upload")
async def upload_file(
    data: UploadFile,
    storage: Storage,
) -> dict:
    """Upload a file and return download URL."""
    content = await data.read()

    result = await storage.put(
        key=f"uploads/{data.filename}",
        data=content,
        content_type=data.content_type,
        metadata={
            "original-name": data.filename,
            "uploaded-by": "user-123",
        },
    )

    download_url = await storage.url(result.key, expires_in=timedelta(hours=24))

    return {
        "key": result.key,
        "size": result.size,
        "content_type": result.content_type,
        "download_url": download_url,
    }

Streaming Large Files¶

from litestar import get
from litestar.response import Stream


@get("/files/{key:path}")
async def stream_file(
    key: str,
    storage: Storage,
) -> Stream:
    """Stream a file directly from S3."""
    info = await storage.info(key)

    return Stream(
        storage.get(key),
        media_type=info.content_type or "application/octet-stream",
        headers={
            "Content-Length": str(info.size),
            "Content-Disposition": f'attachment; filename="{key.split("/")[-1]}"',
        },
    )

Copy Between Prefixes¶

async def publish_draft(key: str, storage: Storage) -> StoredFile:
    """Move a file from drafts to published."""
    source = f"drafts/{key}"
    destination = f"published/{key}"

    result = await storage.copy(source, destination)
    await storage.delete(source)

    return result

IAM and Credential Best Practices¶

Principle of Least Privilege¶

Create IAM policies with minimal required permissions:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "AllowUploadDownload",
            "Effect": "Allow",
            "Action": [
                "s3:PutObject",
                "s3:GetObject"
            ],
            "Resource": "arn:aws:s3:::my-bucket/uploads/*"
        },
        {
            "Sid": "AllowList",
            "Effect": "Allow",
            "Action": "s3:ListBucket",
            "Resource": "arn:aws:s3:::my-bucket",
            "Condition": {
                "StringLike": {
                    "s3:prefix": "uploads/*"
                }
            }
        }
    ]
}

Credential Hierarchy¶

S3Storage uses credentials in this order:

Explicit access_key_id and secret_access_key in config
Environment variables: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY
AWS credentials file (~/.aws/credentials)
IAM role (EC2 instance profile, ECS task role, Lambda execution role)

Recommended approach by environment:

Environment	Credential Method
Local development	Environment variables or credentials file
CI/CD	Environment variables (secrets)
EC2/ECS/Lambda	IAM roles
Kubernetes	IAM Roles for Service Accounts (IRSA)

Environment Variable Configuration¶

# Required
export AWS_ACCESS_KEY_ID="AKIAIOSFODNN7EXAMPLE"
export AWS_SECRET_ACCESS_KEY="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"

# Optional
export AWS_DEFAULT_REGION="us-west-2"
export AWS_ENDPOINT_URL="https://custom-endpoint.com"  # For S3-compatible

Never Commit Credentials¶

# BAD - credentials in code
storage = S3Storage(S3Config(
    bucket="my-bucket",
    access_key_id="AKIAIOSFODNN7EXAMPLE",  # Never do this!
    secret_access_key="wJalrXUtnFEMI/...",
))

# GOOD - credentials from environment
import os

storage = S3Storage(S3Config(
    bucket=os.environ["S3_BUCKET"],
    region=os.environ.get("AWS_REGION", "us-east-1"),
    # Credentials automatically loaded from environment
))

Error Handling¶

from litestar_storages import StorageError, FileNotFoundError


async def safe_download(key: str, storage: Storage) -> bytes | None:
    """Download a file with error handling."""
    try:
        return await storage.get_bytes(key)
    except FileNotFoundError:
        return None
    except StorageError as e:
        logger.error(f"Storage error: {e}")
        raise

Next Steps¶

Learn about Memory storage for testing S3 code without AWS
Create custom backends for other cloud providers