bucket_helper package

Submodules

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:

Usage Example

>>> import bucket_helper as bh
>>> cred = bh.credentials("path/to/s3_config.json")
>>> 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)[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.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.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.exists(s3_address, cred)[source]

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

Parameters:
Return type:

bool

bucket_helper.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.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.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.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.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.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.