Python

Plain Python With boto3

Install

pip install boto3 python-dotenv

Environment

S3_ENDPOINT=https://storage.example.local
S3_ACCESS_KEY=your-access-key
S3_SECRET_KEY=your-secret-key
S3_REGION=us-east-1
S3_BUCKET=my-bucket
S3_USE_PATH_STYLE_ENDPOINT=true

Client

import os
from dotenv import load_dotenv
import boto3
from botocore.config import Config

load_dotenv()

s3 = boto3.client(
    "s3",
    endpoint_url=os.environ["S3_ENDPOINT"],
    aws_access_key_id=os.environ["S3_ACCESS_KEY"],
    aws_secret_access_key=os.environ["S3_SECRET_KEY"],
    region_name=os.getenv("S3_REGION", "us-east-1"),
    config=Config(s3={"addressing_style": "path"}),
)

bucket = os.environ["S3_BUCKET"]

Object Operations

from s3_client import s3, bucket

def upload_object():
    s3.upload_file("avatar.png", bucket, "users/123/avatar.png")

def download_object():
    s3.download_file(bucket, "users/123/avatar.png", "downloaded-avatar.png")

def list_objects():
    response = s3.list_objects_v2(Bucket=bucket, Prefix="users/123/")
    for item in response.get("Contents", []):
        print(item["Key"], item["Size"])

def delete_object():
    s3.delete_object(Bucket=bucket, Key="users/123/avatar.png")

FastAPI

Install

pip install fastapi uvicorn boto3 python-multipart python-dotenv

Upload And Download

import os
import tempfile
from fastapi import FastAPI, UploadFile, HTTPException
from fastapi.responses import StreamingResponse
from s3_client import s3, bucket

app = FastAPI()

@app.post("/upload")
async def upload(file: UploadFile):
    key = f"uploads/{file.filename}"

    with tempfile.NamedTemporaryFile(delete=False) as tmp:
        tmp.write(await file.read())
        tmp_path = tmp.name

    try:
        s3.upload_file(tmp_path, bucket, key, ExtraArgs={"ContentType": file.content_type})
        return {"key": key}
    finally:
        os.unlink(tmp_path)

@app.get("/download/{key:path}")
def download(key: str):
    try:
        obj = s3.get_object(Bucket=bucket, Key=key)
    except Exception as exc:
        raise HTTPException(status_code=404, detail="Object not found") from exc

    return StreamingResponse(
        obj["Body"],
        media_type=obj.get("ContentType", "application/octet-stream"),
        headers={"Content-Disposition": f'attachment; filename="{key.split("/")[-1]}"'},
    )

Run:

uvicorn main:app --reload

Django

Install:

pip install django-storages boto3

Settings example:

INSTALLED_APPS = [
    # ...
    "storages",
]

STORAGES = {
    "default": {
        "BACKEND": "storages.backends.s3boto3.S3Boto3Storage",
        "OPTIONS": {
            "bucket_name": os.environ["S3_BUCKET"],
            "endpoint_url": os.environ["S3_ENDPOINT"],
            "access_key": os.environ["S3_ACCESS_KEY"],
            "secret_key": os.environ["S3_SECRET_KEY"],
            "region_name": os.getenv("S3_REGION", "us-east-1"),
            "addressing_style": "path",
            "default_acl": None,
            "querystring_auth": True,
        },
    },
}

Usage:

from django.core.files.storage import default_storage
from django.core.files.base import ContentFile

default_storage.save("uploads/hello.txt", ContentFile(b"Hello from Django"))

Flask

Install

pip install flask boto3 python-dotenv

Upload And Download

from flask import Flask, request, send_file, abort
from tempfile import NamedTemporaryFile
from s3_client import s3, bucket

app = Flask(__name__)

@app.post("/upload")
def upload():
    file = request.files.get("file")
    if not file:
        abort(400, "Missing file")

    key = f"uploads/{file.filename}"
    s3.upload_fileobj(file.stream, bucket, key, ExtraArgs={"ContentType": file.content_type})
    return {"key": key}

@app.get("/download/<path:key>")
def download(key):
    tmp = NamedTemporaryFile(delete=False)
    try:
        s3.download_fileobj(bucket, key, tmp)
        tmp.flush()
        return send_file(tmp.name, as_attachment=True, download_name=key.split("/")[-1])
    except Exception:
        abort(404)

Common Python Issues

Symptom	Fix
SignatureDoesNotMatch	Use Config(s3={"addressing_style": "path"}) and verify region.
NoCredentialsError	Load .env before creating the boto3 client.
EndpointConnectionError	Use the storage endpoint, not the Quat customer app URL.
Temporary URL fails	Confirm that your deployment supports S3 presigned URLs.