django-capo-s3

A Django file storage backend for S3-compatible object stores, built on the capo-s3 client instead of boto3 — a drop-in alternative to django-storages[s3].

Features

• Media and static storages — S3Storage for media, plus S3StaticStorage (plain) and S3ManifestStaticStorage (content-hashed names + staticfiles.json manifest for cache-busting) for collectstatic. The manifest can live in the bucket or in a separate storage.
• URLs — presigned URLs by default (with per-call expire and response overrides like response_content_disposition), unsigned public URLs, or CloudFront-signed URLs for a custom domain.
• Uploads — streaming single-PUT uploads that automatically switch to a concurrent multipart transfer above a configurable threshold.
• Transparent gzip — eligible content types are stored compressed and decompressed on read.
• Server-side encryption, storage class, cache-control, metadata, ... — via object_parameters, passed straight through to the underlying request.
• Flexible networking options — custom endpoint (e.g. MinIO), path- or virtual-host addressing, TLS verification and custom CA bundles, connection timeouts, pool size, HTTP/HTTPS/SOCKS5 proxies, and retry attempts.
• Test helpers — django_capo_s3.testing ships an in-memory S3 service that speaks the real wire protocol: tests drive the actual backend with no bucket and no network, then assert on the requests it made.
• Fully typed — this is already the bare minimum for new packages.

Installation

uv add django-capo-s3   # or: pip install django-capo-s3

Configuration

Register the backend in Django's STORAGES setting; everything under OPTIONS is passed to the backend. Options not given there fall back to the AWS_* settings (similar to that reads django-storages), so migrating doesn't require moving your configuration in the same change.

STORAGES = {
    "default": {
        "BACKEND": "django_capo_s3.S3Storage",
        "OPTIONS": {
            "bucket": "my-bucket",
            "region": "eu-central-1",
            "location": "media",
        },
    },
    "staticfiles": {
        "BACKEND": "django_capo_s3.S3ManifestStaticStorage",
        "OPTIONS": {"bucket": "my-bucket", "location": "static"},
    },
}

Usage

On a model field

Most apps never touch storage directly. With STORAGES["default"] configured (above), FileField / ImageField just work — uploads, .url, .size, and .open() all go through the backend.

from django.db import models


class Report(models.Model):
    csv = models.FileField(upload_to="reports/")


report = Report.objects.create(csv=uploaded_file)
report.csv.url  # presigned URL to the object
report.csv.size  # size in bytes
report.csv.open().read()  # file contents

Direct storage access

Grab the configured default storage from the registry and use the Storage API directly.

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

storage = storages["default"]
name = storage.save("reports/june.csv", ContentFile(b"col1,col2\n"))  # returns the stored name
storage.exists(name)  # True
storage.size(name)  # size in bytes
with storage.open(name) as f:
    data = f.read()
storage.delete(name)  # no error if it's already gone

Download URLs

url() is presigned by default. Override the lifetime per call, or add response headers — for example, to force a browser "Save as" with a filename.

storage.url("reports/june.csv")  # presigned, default lifetime (url_expire)
storage.url("reports/june.csv", expire=60)  # presigned, valid for 60 seconds
storage.url(
    "reports/june.csv",
    parameters={"response_content_disposition": 'attachment; filename="june.csv"'},
)

For a public bucket, set "querystring_auth": False in OPTIONS to get plain, cacheable URLs instead.

Bulk delete

delete_objects() removes many objects in a single bulk request (per 1000 keys). Missing keys are ignored, just like delete().

storage.delete_objects(["reports/jan.csv", "reports/feb.csv", "reports/mar.csv"])

Other regions and buckets

A storage is bound to one region and bucket, but for_region() and for_bucket() return clones bound to another — cached per value, so repeated calls reuse the same client and connection pool. Handy for routing objects at runtime, for example from a model's FileField.

storage = storages["default"]
storage.for_region("sa-east-1").save("br/report.csv", content)  # store in São Paulo
storage.for_region("ap-southeast-2").save("au/report.csv", content)  # store in Sydney
storage.for_bucket("archive").save("old/report.csv", content)  # same options, another bucket


class Report(models.Model):
    region = models.CharField(max_length=20)
    csv = models.FileField(upload_to="reports/")

    def save(self, *args, **kwargs):
        self.csv.storage = storages["default"].for_region(self.region)
        super().save(*args, **kwargs)

Static files with cache-busting

Point STORAGES["staticfiles"] at S3ManifestStaticStorage. collectstatic then stores each file under a content-hashed name and {% static %} resolves through the manifest, so assets can be served with long-lived caching. Keep the manifest local so web workers don't fetch it from S3 on startup.

from django.core.files.storage import FileSystemStorage

STORAGES["staticfiles"] = {
    "BACKEND": "django_capo_s3.S3ManifestStaticStorage",
    "OPTIONS": {
        "bucket": "my-bucket",
        "location": "static",
        "manifest_storage": FileSystemStorage(location=BASE_DIR / ".static-manifest"),
    },
}

Faster re-deploys. During collectstatic, the hashing pass lists the bucket once and skips uploading any hashed asset whose content is already stored — so an unchanged deploy costs no uploads instead of re-uploading every file. This is on by default ("skip_unchanged": True); set it to False to fall back to Django's behaviour (for example, on an S3-compatible store whose ETag isn't a content MD5).

Serving through a CDN (CloudFront)

Set custom_domain for plain CDN URLs, or add a CloudFront key pair to sign them for a private distribution.

"OPTIONS": {
    "bucket": "my-bucket",
    "custom_domain": "d123.cloudfront.net",
    "cloudfront_key": cloudfront_private_key_pem,  # PEM contents
    "cloudfront_key_id": "K1ABCDEF",
    "url_expire": 300,
}
# storage.url(name) -> https://d123.cloudfront.net/...?Expires=...&Signature=...&Key-Pair-Id=...

Response overrides passed to url() are signed into the CloudFront URL, so they can't be tampered with — configure the distribution to forward them to the S3 origin for them to take effect:

storage.url("report.pdf", parameters={"response_content_disposition": 'attachment; filename="report.pdf"'})

Large uploads and gzip

Uploads switch to a concurrent multipart transfer above multipart_threshold; text assets can be stored gzip-compressed and are transparently decompressed on read.

"OPTIONS": {
    "bucket": "my-bucket",
    "multipart_threshold": 32 * 1024 * 1024,  # start multipart at 32 MiB
    "multipart_chunksize": 16 * 1024 * 1024,
    "multipart_concurrency": 8,               # parts uploaded in parallel
    "gzip": True,                             # compress CSS/JS/JSON/... at rest
}

Content types

Each upload's Content-Type is resolved in this order: an explicit content_type in object_parameters, the type the file object itself reports (Django's uploaded files carry the one the client sent), the guess from the name's extension, then default_content_type. That last step matters for extensionless keys such as statements/<uuid>, which would otherwise be stored with no Content-Type at all.

"OPTIONS": {
    "bucket": "my-bucket",
    "default_content_type": "application/octet-stream",  # the default; used when nothing else is known
}

To store bytes under a type of your choosing, use TypedContentFile — Django's plain ContentFile carries no content type, so there is otherwise nothing for the second step above to read.

from django_capo_s3 import TypedContentFile

storage.save(f"statements/{statement.id}", TypedContentFile(pdf_bytes, content_type="application/pdf"))

Encryption, storage class, and other object metadata

Whatever object_parameters contains is passed straight to each upload — e.g. SSE-KMS plus a storage class and cache header.

"OPTIONS": {
    "bucket": "my-bucket",
    "default_acl": "private",
    "object_parameters": {
        "server_side_encryption": "aws:kms",
        "ssekms_key_id": "arn:aws:kms:eu-central-1:123456789012:key/abcd-...",
        "storage_class": "STANDARD_IA",
        "cache_control": "max-age=86400",
    },
}

Self-hosted stores (MinIO) with a separate public host

When the client reaches the store at an internal endpoint (say http://minio:9000 inside Docker) but browsers must use a different public host, set custom_domain to that public host. With force_path_style the bucket goes into the URL path, matching how path-style S3 serves it.

"OPTIONS": {
    "bucket": "my-bucket",
    "endpoint": "http://minio:9000",   # where the app connects
    "custom_domain": "localhost:9000",  # where browsers connect
    "force_path_style": True,
    "url_protocol": "http",
    "querystring_auth": False,
}
# storage.url("photo.jpg") -> http://localhost:9000/my-bucket/photo.jpg

Networking (endpoint tuning, proxies)

Tune timeouts, the connection pool, retries, TLS verification, and proxies as needed.

"OPTIONS": {
    "bucket": "my-bucket",
    "connect_timeout": 5.0,
    "read_timeout": 30.0,
    "max_connections_per_host": 50,
    "retry_max_attempts": 5,
    "verify": "/etc/ssl/certs/internal-ca.pem",   # or False to disable TLS verification
    "proxies": {"https": "http://proxy.internal:8080"},
}

To swap the HTTP transport itself — say for zapros' Rust backend — name a client builder and the handler that wraps it. The options above are applied to the builder by calling the matching method on it.

from pyreqwest.client import SyncClientBuilder
from zapros import PyreqwestHandler

"OPTIONS": {
    "bucket": "my-bucket",
    "http_client_builder": SyncClientBuilder,  # needs the zapros[pyreqwest] extra
    "http_handler": PyreqwestHandler,
    "connect_timeout": 5.0,
}

Testing

django_capo_s3.testing ships an in-memory S3 service. It speaks the real wire protocol, so a test drives the actual backend — signing, XML, gzip, multipart — with no bucket and no network, and records every request it served.

from django.core.files.base import ContentFile
from django_capo_s3.testing import mock_s3


def test_report_is_uploaded():
    with mock_s3() as s3:
        storage = s3.storage(location="media")

        storage.save("report.csv", ContentFile(b"a,b,c"))

        assert s3["media/report.csv"] == b"a,b,c"
        assert s3.calls.operations == ["PutObject"]
        assert s3.calls.last.headers["content-type"] == "text/csv"

There are pytest fixtures, helpers for taking presigned URLs and SigV4 headers apart, and s3.fail(...) for the error paths. See the testing guide.