# Streaming an upload straight into MinIO / S3

This example streams an HTTP file upload directly into an S3-compatible object store (MinIO) using a {ref}`sharedmemorypipe`. 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 {ref}`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.

```python
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.

```python
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

1. The client `POST`s the file with `Expect: 100-continue` and an `x-object-key` header.
2. 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.
3. The API process writes incoming HTTP body chunks into the pipe; the Worker reads them on the other end.
4. The Worker coalesces chunks into ≥ 5 MiB parts and uploads each to MinIO via multipart upload.
5. 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.
6. 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.

## Related

* {ref}`sharedmemorypipe`
* {doc}`/uploads` — upload lifecycle and the buffered (`Path`) alternative
* {doc}`/examples/worker/shared_memory_pipe` — the mirror-image (Worker → client) example