# MultipartUploader

## class nominal.experimental.ingest.MultipartUploader(timeout, max_part_retries)

Bases: [`object`](https://docs.python.org/3/builtins/functions.html#object "(in Python v3.15)")

Uploads many files via driver-per-file structured concurrency over three pools.

Enqueue files with [`enqueue_file`](https://dev.nominal.io/python/ref/generated/nominal.experimental.ingest.MultipartUploader.md#nominal.experimental.ingest.MultipartUploader.enqueue_file "nominal.experimental.ingest.MultipartUploader.enqueue_file"); track completion via the returned futures. Each file’s
whole lifecycle is collapsed behind a single `Future[str]` that resolves to the object’s
storage location or raises.

Work is split across a small pool (one single-shot `upload_file` task per small-route file), a
driver pool (one sequential driver per multipart file, which is also the in-flight bound on
open multipart uploads), and a part pool (one sign+PUT task per part, submitted only by
running drivers). Neither route can starve the other’s workers in any batch shape.

Wait DAG (deadlock-freedom by inspection): small tasks and part tasks wait only on the
gate; drivers wait on part futures and the gate; the gate waits only on time.

`small_file_route_max_bytes` (EXPERIMENTAL) routes files at/below that size through the
backend single-shot `upload_file` endpoint (one request) rather than multipart, collapsing
the per-file metadata that rate-limits many-small-file batches. Larger files use multipart.

**Tuning.** The knobs compose around one budget and one pipe. Every Nominal API request
(initiate/sign/complete/`upload_file`) competes for the server’s per-caller admission
budget through the `max_nominal_concurrency` lane, while file bytes flow directly to
storage over `max_storage_workers` concurrent PUT streams. Which of the two binds depends
on the batch shape:

- *Many small files* — the admission budget binds long before bandwidth does. The
  single-shot route is the big lever: one request per file instead of multipart’s three.
  Widening the lane does NOT finish the batch sooner once the budget is the wall — it only
  provokes refusals, and a refused single-shot upload re-sends its whole body, so wire
  traffic grows with no time won. The offered request rate is roughly lane width divided
  by request latency; if uploads show heavy throttling (long per-request tails), narrow
  the lane rather than widening it.
- *A few huge files* — bandwidth binds and the lane sits idle. Aggregate throughput is
  roughly streams x per-stream throughput, but only until the network path saturates;
  past saturation, extra streams degrade each other and aggregate can FALL. Find the knee
  empirically: measure at the default `max_storage_workers`, then halve and double it and
  keep whichever is faster. `part_size` (on [`enqueue_file`](https://dev.nominal.io/python/ref/generated/nominal.experimental.ingest.MultipartUploader.md#nominal.experimental.ingest.MultipartUploader.enqueue_file "nominal.experimental.ingest.MultipartUploader.enqueue_file")) sets how many streams one file
  can occupy — keep files at roughly 3+ parts so streams stay fed near file boundaries —
  and `max_files_in_flight` trades a flatter finishing tail (lower) against more
  interleaving (higher).
- *Memory* — each PUT stream holds one full part in memory, so peak buffered memory is
  roughly `max_storage_workers x part_size` (~640 MiB at the defaults). Shrink either to
  fit a constrained host.
- *Unreliable networks* — `file_retry_timeout` (default one hour) rides out outages by
  retrying whole files whose failures look like network weather; [`timeout`](https://dev.nominal.io/python/ref/generated/nominal.experimental.ingest.MultipartUploader.md#nominal.experimental.ingest.MultipartUploader.timeout "nominal.experimental.ingest.MultipartUploader.timeout") and
  [`max_part_retries`](https://dev.nominal.io/python/ref/generated/nominal.experimental.ingest.MultipartUploader.md#nominal.experimental.ingest.MultipartUploader.max_part_retries "nominal.experimental.ingest.MultipartUploader.max_part_retries") bound each hung PUT attempt; a smaller `part_size` makes every retry
  cheaper at the cost of more requests.

When in doubt, keep the defaults — they are benchmark-tuned, and the classic mistake is
raising everything at once, which slows BOTH regimes: streams past saturation contend with
each other, and lane width past the admission budget only converts useful bandwidth into
refused-and-resent requests.

### timeout: float

### max_part_retries: int

### close(*, cancel_pending=False)

Shut the uploader down; afterwards [`enqueue_file`](https://dev.nominal.io/python/ref/generated/nominal.experimental.ingest.MultipartUploader.md#nominal.experimental.ingest.MultipartUploader.enqueue_file "nominal.experimental.ingest.MultipartUploader.enqueue_file") raises.

Either way, every future [`enqueue_file`](https://dev.nominal.io/python/ref/generated/nominal.experimental.ingest.MultipartUploader.md#nominal.experimental.ingest.MultipartUploader.enqueue_file "nominal.experimental.ingest.MultipartUploader.enqueue_file") has returned is settled by the time this
returns — dropped files with `CancelledError` — and is safe to pass to
[`concurrent.futures.wait`](https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.wait "(in Python v3.15)") / `as_completed` from any thread.

Parameters:
:   **cancel_pending** ([*bool*](https://docs.python.org/3/builtins/functions.html#bool "(in Python v3.15)"))

    If false (the default), every enqueued file runs to completion before
    this returns. If true, queued files are dropped and in-flight multipart files are
    cut short at their next part boundary, so the wait is bounded by one round of
    in-flight uploads plus the capped abort pass — not by the rest of the batch.

### classmethod create(client, *, max_storage_workers=None, max_small_file_workers=None, max_files_in_flight=None, small_file_route_max_bytes=DEFAULT_SMALL_FILE_ROUTE_MAX_BYTES, max_nominal_concurrency=NOMINAL_MAX_CONCURRENCY, timeout=30.0, max_part_retries=3, per_request_retry_timeout=DEFAULT_THROTTLE_DEADLINE_S, max_backoff_duration=DEFAULT_MAX_BACKOFF_DURATION_S, file_retry_timeout=DEFAULT_FILE_RETRY_TIMEOUT_S)

Create a MultipartUploader sized for one batch of uploads.

Parameters:
:   - **client** ([*NominalClient*](https://dev.nominal.io/python/ref/generated/nominal.core.NominalClient.md#nominal.core.NominalClient "nominal.core.NominalClient"))

      The `NominalClient` to upload with. Uploads land in the client’s configured
      workspace, and requests ride the client’s shared transport, whose short jittered
      throttle retries are the local layer of the two-layer backoff. The uploader
      never closes anything it borrows from the client.
    - **max_storage_workers** ([*int*](https://docs.python.org/3/builtins/functions.html#int "(in Python v3.15)") *|* *None*)

      Part-pool size — the number of concurrent direct-to-storage PUT
      streams. Defaults to [`DEFAULT_MAX_STORAGE_WORKERS`](https://dev.nominal.io/python/ref/experimental/ingest.md#nominal.experimental.ingest.DEFAULT_MAX_STORAGE_WORKERS "nominal.experimental.ingest.DEFAULT_MAX_STORAGE_WORKERS") (benchmark-tuned). Raising it
      may degrade throughput rather than improve it: past the network path’s
      saturation point, streams contend with each other. Raise it when the path to
      storage is measured to reward more concurrent streams — and tune it together
      with [`enqueue_file`](https://dev.nominal.io/python/ref/generated/nominal.experimental.ingest.MultipartUploader.md#nominal.experimental.ingest.MultipartUploader.enqueue_file "nominal.experimental.ingest.MultipartUploader.enqueue_file")’s `part_size`: each stream holds one full part in memory,
      so peak buffered upload memory is roughly `max_storage_workers × part_size`
      (~640 MiB at the defaults; 40 workers at the default part size is ~2.5 GiB).
    - **max_small_file_workers** ([*int*](https://docs.python.org/3/builtins/functions.html#int "(in Python v3.15)") *|* *None*)

      Small-pool size. Defaults to `max_nominal_concurrency` —
      small files spend their whole task inside the nominal lane, so extra threads
      beyond the lane width would only queue at its entrance.
    - **max_files_in_flight** ([*int*](https://docs.python.org/3/builtins/functions.html#int "(in Python v3.15)") *|* *None*)

      Driver-pool size, which *is* the bound on concurrently open
      multipart uploads. Defaults to half of `max_storage_workers` (rounded up), which
      keeps the part pool fed without piling up a hard-to-drain tail of half-done
      files; there is no unbounded mode.
    - **small_file_route_max_bytes** ([*int*](https://docs.python.org/3/builtins/functions.html#int "(in Python v3.15)") *|* *None*)

      EXPERIMENTAL. Files whose size is <= this many bytes
      are uploaded single-shot via the backend `upload_file` endpoint (one request)
      instead of multipart — avoiding the per-file metadata burst that rate-limits
      many-small-file batches. Defaults to [`DEFAULT_SMALL_FILE_ROUTE_MAX_BYTES`](https://dev.nominal.io/python/ref/experimental/ingest.md#nominal.experimental.ingest.DEFAULT_SMALL_FILE_ROUTE_MAX_BYTES "nominal.experimental.ingest.DEFAULT_SMALL_FILE_ROUTE_MAX_BYTES");
      guarded to [`MAX_SMALL_FILE_ROUTE_BYTES`](https://dev.nominal.io/python/ref/experimental/ingest.md#nominal.experimental.ingest.MAX_SMALL_FILE_ROUTE_BYTES "nominal.experimental.ingest.MAX_SMALL_FILE_ROUTE_BYTES"). Pass None to disable the route and
      send every file through multipart.
    - **max_nominal_concurrency** ([*int*](https://docs.python.org/3/builtins/functions.html#int "(in Python v3.15)"))

      Width of the nominal lane — the maximum number of Nominal
      API requests in flight at once, across every pool. Benchmark-tuned to fill the
      server’s admission budget without provoking refusals; see
      [`NOMINAL_MAX_CONCURRENCY`](https://dev.nominal.io/python/ref/experimental/ingest.md#nominal.experimental.ingest.NOMINAL_MAX_CONCURRENCY "nominal.experimental.ingest.NOMINAL_MAX_CONCURRENCY").
    - **timeout** ([*float*](https://docs.python.org/3/builtins/functions.html#float "(in Python v3.15)"))

      Per-request timeout for direct-to-storage part PUTs.
    - **max_part_retries** ([*int*](https://docs.python.org/3/builtins/functions.html#int "(in Python v3.15)"))

      Attempts per part before the file fails.
    - **per_request_retry_timeout** ([*float*](https://docs.python.org/3/builtins/functions.html#float "(in Python v3.15)"))

      Wall-clock budget, per request, for retrying while the
      server is throttling. Throttling is not a terminal error inside that budget;
      exceeding it raises `NominalRequestThrottledError`. Non-throttle errors always
      fail immediately.
    - **max_backoff_duration** ([*float*](https://docs.python.org/3/builtins/functions.html#float "(in Python v3.15)"))

      Cap on the shared storm damper’s delay. Under sustained
      throttling the damper doubles toward this cap and every retry sleeps a jittered
      fraction of it; successes decay it back to zero.
    - **file_retry_timeout** ([*float*](https://docs.python.org/3/builtins/functions.html#float "(in Python v3.15)") *|* *None*)

      Wall-clock budget, per file, for retrying TRANSIENT failures —
      connection errors, timeouts, throttle-budget exhaustion, HTTP 408/429/5xx —
      by re-running the file’s whole upload after a jittered exponential backoff.
      Sized for network weather: a wifi drop mid-batch pauses the affected files
      rather than failing them. Permanent failures (a broken file, other 4xx,
      unknown errors) never retry and surface immediately regardless of this value.
      Pass None to disable file-level retry.

Returns:
:   An uploader ready to accept files. Close it (or use it as a context manager) to
    release its threads and sessions.

Raises:
:   [**ValueError**](https://docs.python.org/3/builtins/exceptions.html#ValueError "(in Python v3.15)") – A pool size is not positive, or `small_file_route_max_bytes` is outside
    `(0, MAX_SMALL_FILE_ROUTE_BYTES]`.

Return type:
:   *Self*

### enqueue_file(path, *, file_type=None, name=None, part_size=DEFAULT_CHUNK_SIZE)

Schedule a file upload and return a future for its storage location.

Never blocks: the file is validated on the calling thread and then queued on its route’s
pool. The returned future is the executor’s own, so cancelling a file that has not started
yet genuinely dequeues it (and it never initiates anything); cancelling a running file
returns False. The future stays safe to wait on across a cancelling close: [`close`](https://dev.nominal.io/python/ref/generated/nominal.experimental.ingest.MultipartUploader.md#nominal.experimental.ingest.MultipartUploader.close "nominal.experimental.ingest.MultipartUploader.close")
settles every future it drops (see [`close`](https://dev.nominal.io/python/ref/generated/nominal.experimental.ingest.MultipartUploader.md#nominal.experimental.ingest.MultipartUploader.close "nominal.experimental.ingest.MultipartUploader.close")).

Parameters:
:   - **path** ([*Path*](https://docs.python.org/3/library/pathlib.html#pathlib.Path "(in Python v3.15)"))

      File to upload.
    - **file_type** ([*FileType*](https://dev.nominal.io/python/ref/generated/nominal.core.FileType.md#nominal.core.FileType "nominal.core.filetype.FileType") *|* *None*)

      Override the file type inferred from `path`.
    - **name** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)") *|* *None*)

      Override the filename derived from `path`. An existing extension is preserved;
      only names without one use the default extension from `file_type`.
    - **part_size** ([*int*](https://docs.python.org/3/builtins/functions.html#int "(in Python v3.15)"))

      Bytes per multipart part. Ignored on the small-file route. Each in-flight
      part is held fully in memory, so peak buffered memory across the uploader is
      roughly `max_storage_workers × part_size` — tune the two together.

Returns:
:   A future resolving to the uploaded object’s location.

Raises:
:   - [**RuntimeError**](https://docs.python.org/3/builtins/exceptions.html#RuntimeError "(in Python v3.15)") – The uploader is closed.
    - [**FileNotFoundError**](https://docs.python.org/3/builtins/exceptions.html#FileNotFoundError "(in Python v3.15)") – `path` does not exist.
    - [**ValueError**](https://docs.python.org/3/builtins/exceptions.html#ValueError "(in Python v3.15)") – An applied default extension is malformed, the upload name is unsafe for storage,
      or — multipart route only — `part_size`
      is not positive or would need more parts than the storage provider allows. A
      multipart `part_size` below the provider’s non-final-part minimum is permitted
      (deliberately small parts can suit a lossy network) but logged as a warning,
      since the provider may reject the upload at completion.

Return type:
:   *Future*[[*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)")]