Wiring managed object storage (S3/R2) into a Python app (deploymill)

You are an agent updating a deploymill-managed Python app that has just had a managed object-storage bucket provisioned. These env vars are already set in the app's environment and will be present in the container at boot:

• S3_ENDPOINT — the S3-compatible API endpoint (Cloudflare R2 today).
• S3_REGION — the region (auto for R2).
• S3_BUCKET — the bucket name to read/write.
• S3_ACCESS_KEY_ID / S3_SECRET_ACCESS_KEY — bucket-scoped credentials. Treat the secret as sensitive — never log it or echo it back to a client.

Use object storage for large or long-lived blobs: user-uploaded images/video, datasets, anything you serve to clients. Do not use a volume mount for that — a volume is single-host, fixed-size, and can't be served directly. Use Postgres (database) for structured/relational data, not blobs.

Follow these steps in order. Each step is a concrete file edit, not advice.

1. Add the S3 client

Add boto3 to your dependencies (pyproject.toml):

dependencies = [
  # ...existing...
  "boto3>=1.34",
]

boto3 speaks the S3 API, which R2 implements — there is no R2-specific SDK.

2. Create a single client module

Create app/storage.py:

import os
import boto3
from botocore.config import Config

_required = ("S3_ENDPOINT", "S3_BUCKET", "S3_ACCESS_KEY_ID", "S3_SECRET_ACCESS_KEY")
for _k in _required:
    if not os.environ.get(_k):
        raise RuntimeError(
            f'{_k} is not set. Add storage: {{ provider: "r2" }} to '
            ".deploymill/project.json and run reconcile_project."
        )

BUCKET = os.environ["S3_BUCKET"]

s3 = boto3.client(
    "s3",
    endpoint_url=os.environ["S3_ENDPOINT"],
    region_name=os.environ.get("S3_REGION", "auto"),
    aws_access_key_id=os.environ["S3_ACCESS_KEY_ID"],
    aws_secret_access_key=os.environ["S3_SECRET_ACCESS_KEY"],
    # R2 (and most non-AWS S3 endpoints) require path-style addressing.
    config=Config(s3={"addressing_style": "path"}),
)

Build the client once at import time — a module-level singleton — not per request.

3. Upload, fetch, and list objects

from app.storage import s3, BUCKET

def put_object(key: str, body: bytes, content_type: str) -> None:
    s3.put_object(Bucket=BUCKET, Key=key, Body=body, ContentType=content_type)

def get_object(key: str) -> bytes:
    resp = s3.get_object(Bucket=BUCKET, Key=key)
    return resp["Body"].read()

def list_objects(prefix: str = "") -> list[str]:
    resp = s3.list_objects_v2(Bucket=BUCKET, Prefix=prefix)
    return [o["Key"] for o in resp.get("Contents", [])]

4. Serve blobs without leaking credentials

Stream objects back through your own route, or — for large/public media — hand the client a pre-signed GET URL so it fetches directly from the bucket:

def presign_get(key: str, expires_in: int = 300) -> str:
    return s3.generate_presigned_url(
        "get_object", Params={"Bucket": BUCKET, "Key": key}, ExpiresIn=expires_in
    )

What NOT to do

• Don't commit credentials. They live only in the app env, injected by reconcile. Never hardcode S3_* values into the repo.
• Don't park blobs on a volume mount. Volumes are single-host and fixed-size; object storage is the right home for media/datasets.
• Don't rebuild the client per request. One module-level singleton.
• Don't drop the path-style config. R2 needs addressing_style: "path".

After wiring this up, commit via push_files and call deploy.