Storage

UploadKit does not ship storage clients. You implement StorageProvider (sync) or AsyncStorageProvider (async) once — the same classes work for AWS S3 (omit endpoint_url) and MinIO (set endpoint_url) via boto3 / aioboto3.

Install

These are app dependencies, not UploadKit package deps.

Shell

pip install boto3          # sync AWS S3 / MinIO
pip install aioboto3       # async AWS S3 / MinIO

Shell

uv add boto3        # sync
uv add aioboto3     # async

Shell

poetry add boto3    # sync
poetry add aioboto3 # async

Protocols

Sync pipelines call StorageProvider.put(*, bucket, object_name, body, content_type) and expect an etag (or None).

Async pipelines call AsyncStorageProvider.open_write(...), then stream through AsyncObjectWriter: write → complete (or abort on failure).

Examples

Copy these classes into your app. Wire them into Uploader / AsyncUploader as shown on the Core page.

Boto3S3Storage implements StorageProvider. Requires pip install boto3.

s3_sync.py

import boto3
from botocore.client import Config


class Boto3S3Storage:
    """S3-compatible sync storage for AWS S3 or MinIO."""

    def __init__(
        self,
        *,
        access_key: str,
        secret_key: str,
        region: str = "us-east-1",
        endpoint_url: str | None = None,
    ) -> None:
        kwargs: dict = {
            "service_name": "s3",
            "aws_access_key_id": access_key,
            "aws_secret_access_key": secret_key,
            "region_name": region,
            "config": Config(signature_version="s3v4"),
        }
        if endpoint_url:
            kwargs["endpoint_url"] = endpoint_url
        self.client = boto3.client(**kwargs)

    def put(self, *, bucket, object_name, body, content_type):
        resp = self.client.put_object(
            Bucket=bucket,
            Key=object_name,
            Body=body,
            ContentType=content_type,
        )
        return resp.get("ETag")


# AWS S3
storage = Boto3S3Storage(
    access_key="AKIA...",
    secret_key="...",
    region="eu-west-1",
)

# MinIO (local default)
storage = Boto3S3Storage(
    endpoint_url="http://127.0.0.1:9000",
    access_key="minioadmin",
    secret_key="minioadmin",
    region="us-east-1",
)

Multipart streaming writer (5 MiB part size — S3/MinIO rule except the last part). Requires pip install aioboto3. Choosing part size vs AsyncUploader.chunk_size and workers: Performance.

s3_async.py

from __future__ import annotations

import aioboto3
from botocore.client import Config

_PART_SIZE = 5 * 1024 * 1024  # 5 MiB


class AsyncS3Writer:
    def __init__(self, client, *, bucket: str, object_name: str, content_type: str) -> None:
        self._client = client
        self._bucket = bucket
        self._key = object_name
        self._content_type = content_type
        self._upload_id: str | None = None
        self._parts: list[dict] = []
        self._buffer = bytearray()
        self._part_number = 1

    async def _ensure_upload(self) -> None:
        if self._upload_id is not None:
            return
        resp = await self._client.create_multipart_upload(
            Bucket=self._bucket,
            Key=self._key,
            ContentType=self._content_type,
        )
        self._upload_id = resp["UploadId"]

    async def _flush_part(self, data: bytes) -> None:
        await self._ensure_upload()
        assert self._upload_id is not None
        resp = await self._client.upload_part(
            Bucket=self._bucket,
            Key=self._key,
            PartNumber=self._part_number,
            UploadId=self._upload_id,
            Body=data,
        )
        self._parts.append({"ETag": resp["ETag"], "PartNumber": self._part_number})
        self._part_number += 1

    async def write(self, chunk: bytes) -> None:
        self._buffer.extend(chunk)
        while len(self._buffer) >= _PART_SIZE:
            part = bytes(self._buffer[:_PART_SIZE])
            del self._buffer[:_PART_SIZE]
            await self._flush_part(part)

    async def abort(self) -> None:
        if self._upload_id is None:
            return
        await self._client.abort_multipart_upload(
            Bucket=self._bucket,
            Key=self._key,
            UploadId=self._upload_id,
        )
        self._upload_id = None

    async def complete(self) -> str | None:
        if self._buffer:
            await self._flush_part(bytes(self._buffer))
            self._buffer.clear()
        if self._upload_id is None:
            # empty object
            resp = await self._client.put_object(
                Bucket=self._bucket,
                Key=self._key,
                Body=b"",
                ContentType=self._content_type,
            )
            return resp.get("ETag")
        resp = await self._client.complete_multipart_upload(
            Bucket=self._bucket,
            Key=self._key,
            UploadId=self._upload_id,
            MultipartUpload={"Parts": self._parts},
        )
        self._upload_id = None
        return resp.get("ETag")


class AsyncS3Storage:
    """S3-compatible async storage for AWS S3 or MinIO."""

    def __init__(
        self,
        *,
        access_key: str,
        secret_key: str,
        region: str = "us-east-1",
        endpoint_url: str | None = None,
    ) -> None:
        self._session = aioboto3.Session()
        self._client_kwargs: dict = {
            "service_name": "s3",
            "aws_access_key_id": access_key,
            "aws_secret_access_key": secret_key,
            "region_name": region,
            "config": Config(signature_version="s3v4"),
        }
        if endpoint_url:
            self._client_kwargs["endpoint_url"] = endpoint_url
        self._cm = None
        self._client = None

    async def _get_client(self):
        if self._client is None:
            self._cm = self._session.client(**self._client_kwargs)
            self._client = await self._cm.__aenter__()
        return self._client

    async def open_write(self, *, bucket: str, object_name: str, content_type: str):
        client = await self._get_client()
        return AsyncS3Writer(
            client,
            bucket=bucket,
            object_name=object_name,
            content_type=content_type,
        )


# AWS S3
async_storage = AsyncS3Storage(
    access_key="AKIA...",
    secret_key="...",
    region="eu-west-1",
)

# MinIO
async_storage = AsyncS3Storage(
    endpoint_url="http://127.0.0.1:9000",
    access_key="minioadmin",
    secret_key="minioadmin",
)

AWS S3 vs MinIO

Same class for both. Leave endpoint_url unset for AWS S3. For MinIO, set it to your API URL (local default http://127.0.0.1:9000 with minioadmin / minioadmin).

Django settings use the same toggle via AWS_S3_ENDPOINT_URL — see the Django guide. Odoo uses odoo.tools.config keys — see the Odoo guide.

Frameworks

Adapter glue only — storage classes stay in your app:

Django — UPLOADKIT_STORAGE_PROVIDER factory + get_storage_provider()
FastAPI — async streaming or sync via run_sync_upload
aiohttp — store the provider on app["async_storage"]
Odoo — dotted-path factory + optional addon settings

Testing

Use FakeStorageProvider from uploadkit-testing so tests never hit S3 or MinIO.

Shared policy and error conventions: Common patterns. Chunk / part size and workers: Performance. GitHub copy of these samples: Core README.