bucket_helper.api module

Bucket Helper — FastAPI HTTP surface.

Exposes every public function in bucket_helper.main as an HTTP endpoint so bucket-helper can be dropped behind any reverse proxy and consumed by other services. Kept intentionally minimal:

Multipart uploads for the /upload endpoint (UploadFile), streamed straight to a temporary file so large blobs don’t blow up memory.
FileResponse for /download — bytes stream straight from S3 to the client with cleanup after the response is fully sent.
JSON responses for the CRUD reads (/exists, /list, /tempfile, /strip-path).
BackgroundTasks cleans temp files after the response has been streamed — no leftover garbage on disk after a request.

Credentials flow

Two ways to feed credentials to the API:

Server-side default: set BUCKET_HELPER_CONFIG in the process env; every request loads the same credentials via bucket_helper.credentials() (path or folder).
Per-request: send s3_access_key / s3_secret_key / s3_bucket / s3_https (+ optional s3_endpoint_url, s3_region, s3_prefix, s3_use_path_style, s3_verify_ssl) as form fields. Per-request wins over server default.

Install the extra to get the runtime dependencies:

pip install 'bucket-helper[api]'

Then run the app with any ASGI server:

uvicorn bucket_helper.api:app --host 0.0.0.0 --port 8000

Usage Example

>>> # Start the server:
>>> #   uvicorn bucket_helper.api:app --reload
>>> # Probe existence (server-side default cred loaded from BUCKET_HELPER_CONFIG):
>>> #   curl -F 'key=folder/obj.txt' http://localhost:8000/exists
>>> # Upload a file (per-request cred inline, path-style MinIO):
>>> #   curl -F 'file=@in.bin' -F 'key=folder/obj.bin' \
>>> #        -F 's3_access_key=minioadmin' -F 's3_secret_key=minioadmin' \
>>> #        -F 's3_bucket=uploads' -F 's3_https=http://minio.example.com:9000/uploads' \
>>> #        -F 's3_endpoint_url=http://minio.example.com:9000' \
>>> #        -F 's3_use_path_style=true' \
>>> #        http://localhost:8000/upload
>>> # Full OpenAPI docs at http://localhost:8000/docs

Author

Warith Harchaoui, Ph.D. — https://linkedin.com/in/warith-harchaoui/

bucket_helper.api.delete_endpoint(key=fastapi.Form, s3_access_key=fastapi.Form, s3_secret_key=fastapi.Form, s3_bucket=fastapi.Form, s3_https=fastapi.Form, s3_region=fastapi.Form, s3_endpoint_url=fastapi.Form, s3_prefix=fastapi.Form, s3_use_path_style=fastapi.Form, s3_verify_ssl=fastapi.Form)

Delete an S3 object (idempotent).

Parameters:

key (str)
s3_access_key (str | None)
s3_secret_key (str | None)
s3_bucket (str | None)
s3_https (str | None)
s3_region (str | None)
s3_endpoint_url (str | None)
s3_prefix (str | None)
s3_use_path_style (str | None)
s3_verify_ssl (str | None)

Return type:

fastapi.responses.JSONResponse

bucket_helper.api.download_endpoint(background, key=fastapi.Form, s3_access_key=fastapi.Form, s3_secret_key=fastapi.Form, s3_bucket=fastapi.Form, s3_https=fastapi.Form, s3_region=fastapi.Form, s3_endpoint_url=fastapi.Form, s3_prefix=fastapi.Form, s3_use_path_style=fastapi.Form, s3_verify_ssl=fastapi.Form)

Download an S3 object; the file is streamed back and cleaned up after.

Parameters:

background (fastapi.BackgroundTasks)
key (str)
s3_access_key (str | None)
s3_secret_key (str | None)
s3_bucket (str | None)
s3_https (str | None)
s3_region (str | None)
s3_endpoint_url (str | None)
s3_prefix (str | None)
s3_use_path_style (str | None)
s3_verify_ssl (str | None)

bucket_helper.api.exists_endpoint(key=fastapi.Form, s3_access_key=fastapi.Form, s3_secret_key=fastapi.Form, s3_bucket=fastapi.Form, s3_https=fastapi.Form, s3_region=fastapi.Form, s3_endpoint_url=fastapi.Form, s3_prefix=fastapi.Form, s3_use_path_style=fastapi.Form, s3_verify_ssl=fastapi.Form)

Return {"exists": true|false} for the given key.

Parameters:

key (str)
s3_access_key (str | None)
s3_secret_key (str | None)
s3_bucket (str | None)
s3_https (str | None)
s3_region (str | None)
s3_endpoint_url (str | None)
s3_prefix (str | None)
s3_use_path_style (str | None)
s3_verify_ssl (str | None)

Return type:

fastapi.responses.JSONResponse

bucket_helper.api.health()

Simple liveness probe — no dependency check, just proves the app is up.

Return type:: dict

bucket_helper.api.list_endpoint(prefix=fastapi.Form, max_keys=fastapi.Form, s3_access_key=fastapi.Form, s3_secret_key=fastapi.Form, s3_bucket=fastapi.Form, s3_https=fastapi.Form, s3_region=fastapi.Form, s3_endpoint_url=fastapi.Form, s3_prefix=fastapi.Form, s3_use_path_style=fastapi.Form, s3_verify_ssl=fastapi.Form)

List keys under prefix in the default bucket.

Parameters:

prefix (str)
max_keys (int)
s3_access_key (str | None)
s3_secret_key (str | None)
s3_bucket (str | None)
s3_https (str | None)
s3_region (str | None)
s3_endpoint_url (str | None)
s3_prefix (str | None)
s3_use_path_style (str | None)
s3_verify_ssl (str | None)

Return type:

fastapi.responses.JSONResponse

bucket_helper.api.make_bucket_endpoint(bucket=fastapi.Form, s3_access_key=fastapi.Form, s3_secret_key=fastapi.Form, s3_bucket=fastapi.Form, s3_https=fastapi.Form, s3_region=fastapi.Form, s3_endpoint_url=fastapi.Form, s3_prefix=fastapi.Form, s3_use_path_style=fastapi.Form, s3_verify_ssl=fastapi.Form)

Create a bucket. No-op if it already belongs to the caller.

Parameters:

bucket (str)
s3_access_key (str | None)
s3_secret_key (str | None)
s3_bucket (str | None)
s3_https (str | None)
s3_region (str | None)
s3_endpoint_url (str | None)
s3_prefix (str | None)
s3_use_path_style (str | None)
s3_verify_ssl (str | None)

Return type:

fastapi.responses.JSONResponse

bucket_helper.api.strip_path_endpoint(address=fastapi.Form, s3_access_key=fastapi.Form, s3_secret_key=fastapi.Form, s3_bucket=fastapi.Form, s3_https=fastapi.Form, s3_region=fastapi.Form, s3_endpoint_url=fastapi.Form, s3_prefix=fastapi.Form, s3_use_path_style=fastapi.Form, s3_verify_ssl=fastapi.Form)

Return the key part of an s3://bucket/key address.

Parameters:

address (str)
s3_access_key (str | None)
s3_secret_key (str | None)
s3_bucket (str | None)
s3_https (str | None)
s3_region (str | None)
s3_endpoint_url (str | None)
s3_prefix (str | None)
s3_use_path_style (str | None)
s3_verify_ssl (str | None)

Return type:

fastapi.responses.JSONResponse

bucket_helper.api.tempfile_endpoint(ext=fastapi.Form, prefix=fastapi.Form, s3_access_key=fastapi.Form, s3_secret_key=fastapi.Form, s3_bucket=fastapi.Form, s3_https=fastapi.Form, s3_region=fastapi.Form, s3_endpoint_url=fastapi.Form, s3_prefix=fastapi.Form, s3_use_path_style=fastapi.Form, s3_verify_ssl=fastapi.Form)

Return {"s3_address": ..., "public_url": ...} for a fresh random key.

Does NOT upload anything. Since remote_tempfile() deletes the key on exit and we never wrote anything to it, this is a pure name generator — handy when the client wants to know where it will write before doing the upload itself.

Parameters:

ext (str | None)
prefix (str | None)
s3_access_key (str | None)
s3_secret_key (str | None)
s3_bucket (str | None)
s3_https (str | None)
s3_region (str | None)
s3_endpoint_url (str | None)
s3_prefix (str | None)
s3_use_path_style (str | None)
s3_verify_ssl (str | None)

Return type:

fastapi.responses.JSONResponse

bucket_helper.api.upload_endpoint(background, file=fastapi.File, key=fastapi.Form, content_type=fastapi.Form, s3_access_key=fastapi.Form, s3_secret_key=fastapi.Form, s3_bucket=fastapi.Form, s3_https=fastapi.Form, s3_region=fastapi.Form, s3_endpoint_url=fastapi.Form, s3_prefix=fastapi.Form, s3_use_path_style=fastapi.Form, s3_verify_ssl=fastapi.Form)

Upload a local file to S3. Returns the resulting s3://bucket/key URI.

Parameters:

background (fastapi.BackgroundTasks)
file (fastapi.UploadFile)
key (str | None)
content_type (str | None)
s3_access_key (str | None)
s3_secret_key (str | None)
s3_bucket (str | None)
s3_https (str | None)
s3_region (str | None)
s3_endpoint_url (str | None)
s3_prefix (str | None)
s3_use_path_style (str | None)
s3_verify_ssl (str | None)

Return type:

fastapi.responses.JSONResponse