bucket_helper package

Submodules

• bucket_helper.api module
  • Credentials flow
  • Usage Example
  • Author
  • delete_endpoint()
  • download_endpoint()
  • exists_endpoint()
  • health()
  • list_endpoint()
  • make_bucket_endpoint()
  • strip_path_endpoint()
  • tempfile_endpoint()
  • upload_endpoint()
• bucket_helper.cli module
  • Design notes
  • Usage Example
  • Author
  • main()
• bucket_helper.main module
  • S3-compatible endpoints
  • Usage Example
  • Author
  • credentials()
  • delete()
  • download()
  • exists()
  • get_client_s3()
  • list_prefix()
  • list_prefix_page()
  • make_bucket()
  • remote_tempfile()
  • strip_s3_path()
  • upload()
• bucket_helper.mcp module
  • Author
  • main()

Module contents

Bucket Helper

Utility functions for AWS S3 and any S3-compatible object storage (MinIO, Backblaze B2 S3 API, DigitalOcean Spaces, Cloudflare R2, Wasabi, …) via boto3. Same shape as sftp-helper: credentials() loader + upload / download / delete / exists / list_prefix / make_bucket + a remote_tempfile context manager for stage-and-share flows.

Also exposes multi-surface entry points:

• Library (this module) — the pure functions.

• CLI — bucket_helper.cli (bucket-helper), built with click.

• FastAPI HTTP — bucket_helper.api (uvicorn bucket_helper.api:app).

Usage Example

>>> import bucket_helper as bh
>>> cred = bh.credentials("path/to/settings.yaml")
>>> uri = bh.upload("local.txt", cred, "folder/uploaded.txt")
>>> assert bh.exists(uri, cred)
>>> bh.download(uri, "back.txt", cred)
>>> bh.delete(uri, cred)

Author

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

bucket_helper.credentials(config_path=None)

Retrieve S3 credentials from a configuration file, folder, or environment.

Loads (in this precedence order) from JSON / YAML files, .env, then environment variables — see os_helper.get_config().

Required keys

• s3_access_key : your access key ID

• s3_secret_key : your secret access key

• s3_bucket : default bucket name

• s3_https : base public URL for built objects, e.g. "https://my-bucket.s3.eu-west-3.amazonaws.com" or "https://cdn.example.com" (used by remote_tempfile to build the public URL handed back to the caller)

Optional keys

• s3_region : AWS region (defaults to "us-east-1" if absent, mostly irrelevant for path-style MinIO etc.)

• s3_endpoint_url : custom endpoint for S3-compatible storage (MinIO, R2, B2, Spaces, Wasabi). Empty / missing → AWS S3.

• s3_prefix : default key prefix (path-like, e.g. "uploads")

• s3_use_path_style : "true" to force path-style addressing (typical for MinIO with custom domains). Default "false" (uses virtual-hosted style).

• s3_verify_ssl : "false" to disable TLS verification (use sparingly — e.g. dev MinIO with self-signed cert).

returns:

Credentials dict.

rtype:

dict

Parameters:

config_path (str | None)

Return type:

dict

bucket_helper.delete(s3_address, cred)

Delete an S3 object. Returns True if the object is gone after the call (including the case where it never existed — S3’s delete is idempotent).

Parameters:

• s3_address (str)

• cred (dict)

Return type:

bool

bucket_helper.download(s3_address, local_path, cred)

Download an S3 object to a local path. Returns local_path on success.

Parameters:

• s3_address (str)

• local_path (str)

• cred (dict)

Return type:

str

bucket_helper.exists(s3_address, cred)

Return True if the object at s3_address exists, False otherwise.

Parameters:

• s3_address (str)

• cred (dict)

Return type:

bool

bucket_helper.get_client_s3(cred)

Open a boto3 S3 client honoring cred configuration.

The client is configured per AWS S3 (no endpoint_url) or per the S3-compatible endpoint when cred["s3_endpoint_url"] is set.

Yields:

boto3.client – Ready-to-use S3 client.

Parameters:

cred (dict)

Return type:

Iterator[boto3.client]

bucket_helper.list_prefix(prefix, cred, *, max_keys=1000)

List the object keys under prefix in the default bucket.

Returns at most max_keys keys, ordered by S3’s lexicographic response (no client-side sort). For very large prefixes, paginate via boto3 directly with get_client_s3.

Parameters:

• prefix (str)

• cred (dict)

• max_keys (int)

Return type:

list[str]

bucket_helper.list_prefix_page(prefix, cred, *, max_keys=1000)

List keys under prefix, and say whether more were left behind.

list_prefix() returns at most max_keys keys and cannot tell the caller whether that was all of them. For a human reading a terminal that is a mild annoyance; for a program it is a correctness problem, because a response of 1000 keys looks identical whether the prefix holds exactly 1000 objects or fifty thousand.

This asks S3 for one key more than it will return. If that extra key arrives, there is more behind it. The cost is a single additional key, and it turns an ambiguous answer into a definite one.

Parameters:

• prefix (str) – Key prefix to list under, in the credentials’ default bucket.

• cred (dict) – Credentials, as accepted by get_client_s3().

• max_keys (int, default 1000) – Maximum number of keys to return.

Returns:

The keys (at most max_keys of them), and True when S3 held more than that under the prefix.

Return type:

tuple[list[str], bool]

Examples

>>> keys, truncated = list_prefix_page("logs/", cred, max_keys=10)
>>> if truncated:
...     print("there are more than 10 objects under logs/")

bucket_helper.make_bucket(bucket, cred)

Create a bucket if it doesn’t exist.

Honors cred["s3_region"] for the LocationConstraint. No-op when the bucket already exists and is owned by the credentials.

Parameters:

• bucket (str)

• cred (dict)

Return type:

None

bucket_helper.remote_tempfile(cred, *, ext='', prefix='')

Yield (s3_address, public_url) for a unique key, deleted on exit.

Useful for stage-and-share flows: upload a generated file, hand the public URL to whoever needs it (a downstream worker, a webhook target, a test fixture), and let the context manager clean up the remote artifact at the end of the block — even on exception.

Parameters:

• cred (dict) – Credentials dict from credentials(). The object goes under cred["s3_bucket"]; the public URL is built from cred["s3_https"].

• ext (str, optional) – Extension to append to the random name (with or without leading .). Default empty.

• prefix (str, optional) – Extra path segment under the bucket / default prefix. Default empty.

Yields:

(s3_address, public_url) (tuple of str) – s3_address is "s3://bucket/key"; public_url is the https://... URL built from cred["s3_https"].

Return type:

Iterator[tuple[str, str]]

Example

>>> import bucket_helper as bh
>>> cred = bh.credentials("path/to/settings.yaml")
>>> with bh.remote_tempfile(cred, ext="json") as (addr, url):
...     bh.upload("payload.json", cred, addr, content_type="application/json")
...     # publish `url` somewhere, e.g. trigger a webhook
>>> # the object is gone after the block — even if the body raised.

bucket_helper.strip_s3_path(s3_address, cred)

Return the key part of s3_address (compat with sftp-helper’s strip_sftp_path).

Parameters:

• s3_address (str)

• cred (dict)

Return type:

str

bucket_helper.upload(local_path, cred, s3_address='', content_type=None)

Upload a local file to S3.

Parameters:

• local_path (str) – Path to the local file.

• cred (dict) – Credentials dict from credentials().

• s3_address (str, optional) – Destination — either "s3://bucket/key" or a key under the default bucket. If empty, a random unique hex name is generated (not derived from the file’s content) under cred["s3_prefix"] (if set) of the default bucket.

• content_type (str, optional) – Override the MIME Content-Type header. If None, boto3 lets the server default (typically application/octet-stream).

Returns:

The full s3://bucket/key URI of the uploaded object.

Return type:

str

Raises:

• botocore.exceptions.ClientError – If the upload fails (permissions, bucket missing, network).

• AssertionError – If local_path does not exist or is empty.