Streaming an upload straight into MinIO / S3¶

This example streams an HTTP file upload directly into an S3-compatible object store (MinIO) using a SharedMemoryPipe in Workers. The file is forwarded client → Worker → bucket in chunks: it is never fully held in memory and never staged on local disk.

When to use this¶

You accept large file uploads and want them in object storage, not on the app server's disk.
You do not want to buffer the whole upload before forwarding it.
You want end-to-end backpressure: if the bucket (or network to it) is slow, the HTTP upload is throttled automatically rather than piling up in memory.

This relies on the streamed-upload path: the event must be file-only and the client must send Expect: 100-continue. See The other direction: reading an upload as it arrives. Without 100-continue the body is buffered to a temporary file and File.path is a Path instead.

Event and response definitions¶

The event is file-only: its only non-injected field is typed File. The target object key is taken from a header (header/cookie/scope fields are ignored when deciding whether an event is file-only), so the event stays eligible for the streaming path.

from http import HTTPMethod

from edri.api.dataclass import File, Header
from edri.api.dataclass.api_event import api
from edri.dataclass.event import Event
from edri.dataclass.response import Response, response


@response
class UploadResponse(Response):
    bucket: str
    key: str
    etag: str


@api(url="/upload", method=HTTPMethod.POST)
class UploadToBucket(Event):
    file: File
    object_key: list[str] | None = Header(name="x-object-key")
    response: UploadResponse

Worker example¶

S3 multipart parts must be at least 5 MiB (except the final part), while the pipe hands us smaller chunks (EDRI_CHUNK_SIZE). The Worker therefore coalesces chunks into ≥ 5 MiB buffers and uploads one part per buffer.

import boto3
from botocore.config import Config

from edri.abstract.worker import WorkerProcess
from edri.utility.shared_memory_pipe import SharedMemoryPipe

from events import UploadToBucket

BUCKET = "uploads"
MIN_PART = 5 * 1024 * 1024  # S3 minimum part size (except the last part)


def make_client():
    return boto3.client(
        "s3",
        endpoint_url="http://minio:9000",
        aws_access_key_id="minioadmin",
        aws_secret_access_key="minioadmin",
        config=Config(signature_version="s3v4"),
    )


class UploadToBucketWorker(WorkerProcess[UploadToBucket]):

    def do(self) -> None:
        source = self.event.file.path           # a SharedMemoryPipe reader
        if not isinstance(source, SharedMemoryPipe):
            # No 100-continue: framework buffered to disk; fall back to a plain read.
            with open(source, "rb") as fh:
                self._upload(fh.read())
            return

        client = make_client()
        key = self.event.object_key[0]

        upload_id = client.create_multipart_upload(Bucket=BUCKET, Key=key)["UploadId"]
        parts: list[dict] = []
        buffer = bytearray()
        part_number = 1

        try:
            for chunk in source:                # blocks (backpressure) if MinIO is slow
                buffer += chunk
                if len(buffer) >= MIN_PART:
                    parts.append(self._put_part(client, key, upload_id, part_number, bytes(buffer)))
                    part_number += 1
                    buffer.clear()

            # flush the remainder as the final part (may be < 5 MiB, which is allowed)
            if buffer or not parts:
                parts.append(self._put_part(client, key, upload_id, part_number, bytes(buffer)))

            result = client.complete_multipart_upload(
                Bucket=BUCKET, Key=key, UploadId=upload_id,
                MultipartUpload={"Parts": parts},
            )
        except Exception:
            client.abort_multipart_upload(Bucket=BUCKET, Key=key, UploadId=upload_id)
            raise

        self.event.response.bucket = BUCKET
        self.event.response.key = key
        self.event.response.etag = result["ETag"]
        self.event_send(self.event)

    @staticmethod
    def _put_part(client, key, upload_id, number, data):
        result = client.upload_part(
            Bucket=BUCKET, Key=key, UploadId=upload_id,
            PartNumber=number, Body=data,
        )
        return {"ETag": result["ETag"], "PartNumber": number}

How it works¶

The client POSTs the file with Expect: 100-continue and an x-object-key header.
EDRI sees a file-only event, creates a SharedMemoryPipe, sets event.file.path to its reader, and dispatches the event immediately — before the upload completes.
The API process writes incoming HTTP body chunks into the pipe; the Worker reads them on the other end.
The Worker coalesces chunks into ≥ 5 MiB parts and uploads each to MinIO via multipart upload.
If the Worker (or MinIO) reads slowly, the pipe fills, the API's write blocks, and the HTTP receive is throttled — backpressure all the way to the client.
On completion the Worker sends the event back with the bucket, key, and ETag; on any error it aborts the multipart upload so no partial object is left behind.

Requirements¶

A running MinIO (or any S3-compatible store) reachable at endpoint_url, with the target bucket created.
boto3 installed in the Worker's environment.
Clients must send Expect: 100-continue and a base64-encoded filename header for the streaming path to engage.