bucket_helper.main module

Bucket Helper

Utility functions to interact with AWS S3 and any S3-compatible object storage: upload, download, list, exists, remove, plus a remote_tempfile context manager that gives you a unique key and cleans up automatically.

Backed by boto3. SSL certificate verification is on by default and never disabled silently — callers may opt out per-call via cred["s3_verify_ssl"] = false only if they really mean it (handy for a self-signed MinIO behind a corporate VPN).

S3-compatible endpoints

Anything that speaks the AWS S3 API works transparently — just set cred["s3_endpoint_url"] to the endpoint:

  • AWS S3 : leave s3_endpoint_url empty or unset (boto3 picks s3.<region>.amazonaws.com automatically).

  • MinIO : "http://minio.example.com:9000" or "https://minio.example.com".

  • DigitalOcean Spaces : "https://nyc3.digitaloceanspaces.com".

  • Cloudflare R2 : "https://<account_id>.r2.cloudflarestorage.com".

  • Backblaze B2 (S3 API) : "https://s3.us-west-001.backblazeb2.com".

  • Wasabi : "https://s3.us-east-1.wasabisys.com".

Usage Example

>>> import bucket_helper as bh
>>> cred = bh.credentials("path/to/s3_config.json")
>>> # Upload a local file at an explicit key.
>>> uri = bh.upload("local.bin", cred, "folder/local.bin")
>>> # Probe it, download it, list around it, then delete it.
>>> assert bh.exists(uri, cred)
>>> bh.download(uri, "back.bin", cred)
>>> [k for k in bh.list_prefix("folder/", cred)]  # noqa: E501
>>> bh.delete(uri, cred)

Author

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

bucket_helper.main.credentials(config_path=None)[source]

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.main.delete(s3_address, cred)[source]

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:
Return type:

bool

bucket_helper.main.download(s3_address, local_path, cred)[source]

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

Parameters:
Return type:

str

bucket_helper.main.exists(s3_address, cred)[source]

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

Parameters:
Return type:

bool

bucket_helper.main.get_client_s3(cred)[source]

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.main.list_prefix(prefix, cred, *, max_keys=1000)[source]

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:
Return type:

list[str]

bucket_helper.main.make_bucket(bucket, cred)[source]

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:
Return type:

None

bucket_helper.main.remote_tempfile(cred, *, ext='', prefix='')[source]

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/s3_config.json")
>>> 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.main.strip_s3_path(s3_address, cred)[source]

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

Parameters:
Return type:

str

bucket_helper.main.upload(local_path, cred, s3_address='', content_type=None)[source]

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 content-hashed name is generated 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.