remote-upload

Stream local content INTO remote storage (S3 / MinIO, Azure Blob, GCS, SFTP, authenticated HTTP) through one tiny, framework-agnostic API — the write-side twin of remote-download.

from remote_upload import RemoteUpload

result = (
    RemoteUpload.to(target)
    .body(stream, length)
    .content_type("image/jpeg")
    .upload()
)

remote-download pipes bytes out of a remote origin to your client. remote-upload is the other half: it pushes bytes into a remote destination. Same shape, mirrored — works in any web framework (Django, FastAPI, Flask), task queues, AWS Lambda, or plain CLI scripts: anywhere you can read bytes from a stream.

Install

The core (HttpTarget) is pure standard library — no third-party dependencies. Cloud and SSH backends each pull in one SDK, gated behind an extra so you install only what you use:

pip install remote-upload

Targets are importable straight from the package root. The extra-gated ones are loaded lazily, so importing one without its SDK installed raises a clear ImportError telling you exactly which extra to install:

from remote_upload import RemoteUpload, S3Target, AzureBlobTarget  # lazily resolved

Requires Python 3.14+.

Quick start

Every upload follows the same fluent shape: pick a target (or a URL), attach a body, optionally decorate it, then call .upload().

Plain HTTP PUT to a URL

A bare string is treated as an absolute HTTP/HTTPS URL and wrapped in a default HttpTarget (PUT, no auth):

from remote_upload import RemoteUpload

result = (
    RemoteUpload.to("https://api.example.com/files/report.pdf")
    .body(b"%PDF-1.7 ...")
    .content_type("application/pdf")
    .upload()
)
print(result.key, result.bytes_transferred, "bytes")

The three ways to supply a body

# 1. Raw bytes — content length is exact and inferred for you.
RemoteUpload.to(target).body(b"hello world").upload()

# 2. An open binary stream — pass length when you know it (cloud targets like
#    S3 need it); omit it for chunked / unknown-length uploads.
with open("photo.jpg", "rb") as fh:
    RemoteUpload.to(target).body(fh, length=204_800).upload()

# 3. A file on disk — body_file() infers length, and (unless already set) the
#    filename and content type from the path.
RemoteUpload.to(target).body_file("/tmp/photo.jpg").upload()

upload() consumes and closes the body stream. Build a fresh request per upload — instances are not reusable.

Everything together

def on_progress(sent: int, total: int | None) -> None:
    pct = (sent * 100 // total) if total else -1
    print(f"uploaded {sent} / {total} bytes ({pct}%)")

result = (
    RemoteUpload.to(target)
    .body(stream, length=204_800)
    .content_type("image/jpeg")
    .metadata({"captured_by": "user-1", "album": "summer"})
    .checksum("sha256")            # also accepts Java-style "SHA-256"
    .on_progress(on_progress)
    .upload()
)

print(result.key)
print(result.etag)
print(result.checksum_hex)
print(f"{result.bytes_per_second / 1_048_576:.1f} MiB/s")

Backends

Every target is constructed with keyword arguments and then handed to RemoteUpload.to(...). The keyword names below match each target's constructor exactly.

HttpTarget — authenticated HTTP PUT (stdlib, always available)

from remote_upload import RemoteUpload, HttpTarget

target = HttpTarget(
    "https://api.example.com/files/report.pdf",
    method="PUT",
    headers={"X-Api-Key": "secret"},
    bearer="<token>",            # adds "Authorization: Bearer <token>"
    connect_timeout=10.0,
    request_timeout=60.0,
)
RemoteUpload.to(target).body_file("/tmp/report.pdf").upload()

S3Target — S3 / MinIO / Ceph / LocalStack ([s3])

from remote_upload import RemoteUpload, S3Target

target = S3Target(
    bucket="my-bucket",
    key="tenant-1/uploads/abc/photo.jpg",
    endpoint="http://localhost:9000",   # MinIO; omit for real AWS
    access_key="minioadmin",
    secret_key="minioadmin",
    region="us-east-1",
)

result = (
    RemoteUpload.to(target)
    .body(stream, length=204_800)
    .content_type("image/jpeg")
    .metadata({"captured_by": "user-1"})
    .checksum("sha256")
    .upload()
)
print(result.key, "etag=", result.etag, result.bytes_transferred, "bytes")

Setting endpoint enables path-style addressing automatically (needed by most S3-compatible services); override with path_style=... if needed. Omit access_key / secret_key to fall back to the default boto3 credential chain (env vars, ~/.aws/credentials, IAM roles). For high throughput inject a shared boto3 client with client=... — an injected client is reused and never closed by the target.

AzureBlobTarget — Azure Blob Storage ([azure])

from remote_upload import RemoteUpload, AzureBlobTarget

target = AzureBlobTarget(
    container="uploads",
    blob="tenant-1/photo.jpg",
    connection_string="DefaultEndpointsProtocol=https;AccountName=...;AccountKey=...",
)
RemoteUpload.to(target).body_file("/tmp/photo.jpg").content_type("image/jpeg").upload()

Authenticate one of three ways: a full connection_string, an endpoint URL plus an optional sas_token, or a pre-built BlobClient passed as client=.... Uploads overwrite an existing blob.

GcsTarget — Google Cloud Storage ([gcs])

from remote_upload import RemoteUpload, GcsTarget

target = GcsTarget(
    bucket="my-bucket",
    object_name="tenant-1/photo.jpg",
    project_id="my-gcp-project",
    credentials_path="/etc/secrets/service-account.json",
)
RemoteUpload.to(target).body_file("/tmp/photo.jpg").content_type("image/jpeg").upload()

Credentials resolve in order: an explicit credentials object, then a service-account JSON file via credentials_path (an optional file: prefix is stripped), then Application Default Credentials. Pass a pre-built storage.Client via client=... for reuse / tests — an injected client is never closed by the target.

SftpTarget — SFTP over SSH ([sftp])

from remote_upload import RemoteUpload, SftpTarget

target = SftpTarget(
    host="sftp.example.com",
    user="deploy",
    path="/uploads/photo.jpg",
    port=22,
    password="s3cr3t",                       # or use private_key_path=...
)
RemoteUpload.to(target).body_file("/tmp/photo.jpg").upload()

Authenticate with a password or a private_key_path (Ed25519 / ECDSA / RSA / DSA are tried in order). Authentication failures surface as TerminalUploadError; connection and I/O failures as RetryableUploadError. Tune connect_timeout / auth_timeout as needed.

HttpxTarget — HTTP PUT with retries / auth / proxy ([httpx])

from remote_upload import RemoteUpload, HttpxTarget

target = HttpxTarget(
    "https://api.example.com/files/report.pdf",
    method="PUT",
    bearer="<token>",                        # or basic_auth=("user", "pass")
    retries=3,
    connect_timeout=30.0,
    response_timeout=300.0,
    proxy="http://corp-proxy:3128",
)
RemoteUpload.to(target).body_file("/tmp/report.pdf").upload()

The richer twin of the stdlib HttpTarget: transport-level retries, Bearer / Basic auth, granular timeouts and an optional forward proxy. Pass an existing httpx.Client via client=... to reuse a connection pool.

Framework integrations

Optional one-line adapters take the upload object your web framework parsed and stream it straight into a target — carrying its content type and filename. Each lives behind its own extra:

pip install "remote-upload[fastapi]"   # or [flask] / [django]

# FastAPI / Starlette  — use a sync `def` so the transfer runs in the threadpool
from fastapi import UploadFile
from remote_upload import AzureBlobTarget
from remote_upload.integrations.fastapi import upload

@app.post("/uploads")
def create(file: UploadFile):
    target = AzureBlobTarget(container="media", blob=file.filename, connection_string=CONN)
    result = upload(target, file)
    return {"key": result.key, "bytes": result.bytes_transferred, "etag": result.etag}

# Flask
from flask import request
from remote_upload.integrations.flask import upload

@app.post("/uploads")
def create():
    file = request.files["file"]
    target = AzureBlobTarget(container="media", blob=file.filename, connection_string=CONN)
    result = upload(target, file)
    return {"key": result.key, "bytes": result.bytes_transferred}

# Django
from remote_upload.integrations.django import upload

def create(request):
    file = request.FILES["file"]
    target = AzureBlobTarget(container="media", blob=file.name, connection_string=CONN)
    result = upload(target, file)
    return JsonResponse({"key": result.key, "bytes": result.bytes_transferred})

Not on this list (plain WSGI/ASGI, a CLI, a task queue)? You need no adapter — use the core RemoteUpload.to(target).body(stream, length).upload() directly. Under the hood every adapter is a thin wrapper over the framework-neutral remote_upload.integrations._base.upload(target, stream=..., length=..., content_type=..., filename=...).

Concepts

The library is built around a single port (UploadTarget) and three plain data types. Implement the port and you can push bytes to anything.

UploadTarget — the port

A Protocol with one method:

def upload(self, content: UploadContent) -> UploadResult: ...

Each backend supplies its own implementation; consumers push bytes through the same API regardless of where they land. Custom destinations only need this single method. Implementations read content.body but do not own its lifecycle — the request opens and closes the stream for them.

UploadContent — the payload

A frozen dataclass the facade builds and hands to the target:

UploadResult — the outcome

A frozen dataclass combining the target's provider identifiers with the transfer stats the request measures:

ProgressListener — progress callback

A callable (bytes_transferred: int, total_bytes: int | None) -> None, fired as the destination reads the body. total_bytes is None for chunked / unknown-length uploads. Register it with .on_progress(...):

RemoteUpload.to(target).body(data).on_progress(
    lambda sent, total: print(f"{sent}/{total}")
).upload()

Error handling — retryable vs terminal

Targets translate provider failures into one of two exceptions so callers can branch on retry semantics without parsing messages. Both subclass RemoteUploadError:

• RetryableUploadError — transient: a network blip, a 5xx response, a timeout. Callers with a retry budget (an offline outbox, a sync coordinator) should re-enqueue with backoff.
• TerminalUploadError — permanent: invalid credentials, a 4xx, quota exceeded, validation. Retrying the same request will fail again; change something (re-auth, fix the payload, escalate) instead.

from remote_upload import (
    RemoteUpload,
    RetryableUploadError,
    TerminalUploadError,
)

try:
    RemoteUpload.to(target).body_file("/tmp/photo.jpg").upload()
except RetryableUploadError:
    enqueue_for_retry(...)        # backoff and try again later
except TerminalUploadError:
    mark_failed_and_alert(...)    # do not retry; surface to the user

This retryable/terminal split is the deliberate improvement over a single exception type: it lets a sync coordinator decide between "keep retrying" and "mark failed, surface to user".

Java -> Python mapping

This package is a faithful port of the Java library remote-upload-java. If you know one, you know the other:

In short: *Exception becomes *Error, fluent builders become keyword arguments, and getters become attributes.

License

MIT (c) Carlos Guillermo Reyes Ramiro. See LICENSE.