"""
Bucket Helper — click-based command-line interface.
The one command-line surface, installed as the ``bucket-helper`` entry
point. It used to be the optional twin of an argparse CLI, kept in
lock-step by hand and reachable only under a ``-click`` suffix that named
the parsing library rather than the interface; both are gone.
Design notes
------------
- Flags keep the names the argparse door used (``--input`` / ``--key`` /
…) rather than the more idiomatic click positional style: a spelling
people already type is worth more than micro-idiomaticity.
- Subcommand names are the ones the FastAPI surface introspects, so the
two doors cannot answer differently.
- A library exception surfaces as a clean ``Error: ...`` line + exit 1
(see :func:`main`), not a raw traceback; click's own control flow
(usage errors, ``--help``) is untouched.
Usage Example
-------------
>>> # bucket-helper upload --config settings.yaml --input local.txt --key folder/uploaded.txt
>>> # bucket-helper download --config settings.yaml --key folder/uploaded.txt --output local.txt
>>> # bucket-helper list --config settings.yaml --prefix folder/
>>> # bucket-helper tempfile --config settings.yaml --ext json --prefix runs
Author
------
Warith Harchaoui, Ph.D. — https://linkedin.com/in/warith-harchaoui/
"""
from __future__ import annotations
import json
import sys
from collections.abc import Callable
from typing import Any
import click
# The same functions the library and the HTTP surface call — one source of truth.
from . import (
credentials,
delete,
download,
exists,
list_prefix_page,
make_bucket,
remote_tempfile,
strip_s3_path,
upload,
)
# ---------------------------------------------------------------------------
# Top-level group
#
# ``invoke_without_command=False`` forces the user to name a subcommand;
# ``context_settings`` widens the help output so long option lists stay
# readable on modern terminals.
# ---------------------------------------------------------------------------
@click.group(
context_settings={"help_option_names": ["-h", "--help"], "max_content_width": 100},
)
# ``-V`` beside ``--version``, as the argparse door offered. The version comes
# from the installed distribution metadata because ``pyproject.toml`` is this
# package's only version site — there is no ``__version__`` to drift from it.
@click.version_option(None, "-V", "--version", prog_name="bucket-helper")
def cli() -> None:
"""Bucket Helper — S3-compatible object storage from the shell."""
# Nothing to do at the group level — every subcommand carries its
# own arguments and side effects.
# ---------------------------------------------------------------------------
# Shared option: --config
# ---------------------------------------------------------------------------
# Repeating the ``--config`` option across eight subcommands would be
# noisy — we define it once as a decorator and apply it everywhere.
def _config_option(f: Callable[..., Any]) -> Callable[..., Any]:
"""Attach the shared ``--config`` click option to a command callback.
Parameters
----------
f : callable
The click command callback to decorate.
Returns
-------
callable
The same callback with the ``--config`` option registered.
"""
# Applying the decorator programmatically lets every subcommand share
# one definition of ``--config`` instead of repeating it eight times.
return click.option(
"--config",
default=None,
type=click.Path(exists=False),
help="Path to settings.yaml (or .json), or a folder holding one. Empty = env-only.",
)(f)
# ---------------------------------------------------------------------------
# upload
#
# We name each click command explicitly (``cli.command("verb")``) so the
# subcommand name is stable regardless of the Python function name, which
# has to be suffixed (``upload_``, ``list_cmd`` …) to avoid shadowing the
# library-level import of the same verb (``from . import upload``).
# ---------------------------------------------------------------------------
@cli.command("upload")
@_config_option
@click.option(
"--input", "input_", required=True, type=click.Path(exists=True), help="Local file path."
)
@click.option(
"--key", default=None, help="Destination key or full 's3://bucket/key' URI (empty = auto)."
)
@click.option(
"--content-type",
default=None,
help="Override the S3 Content-Type header (e.g. application/json).",
)
def upload_(config: str | None, input_: str, key: str | None, content_type: str | None) -> None:
"""Upload a local file to S3."""
# Thin dispatch to the library.
cred = credentials(config)
uri = upload(local_path=input_, cred=cred, s3_address=key or "", content_type=content_type)
click.echo(uri)
# ---------------------------------------------------------------------------
# download
# ---------------------------------------------------------------------------
@cli.command("download")
@_config_option
@click.option("--key", required=True, help="Source key or full 's3://bucket/key' URI.")
@click.option("--output", required=True, type=click.Path(), help="Destination local file path.")
def download_(config: str | None, key: str, output: str) -> None:
"""Download an S3 object to a local path."""
cred = credentials(config)
out = download(s3_address=key, local_path=output, cred=cred)
click.echo(out)
# ---------------------------------------------------------------------------
# delete
# ---------------------------------------------------------------------------
@cli.command("delete")
@_config_option
@click.option("--key", required=True, help="Object address — 's3://bucket/key' or bare key.")
def delete_(config: str | None, key: str) -> None:
"""Delete an S3 object (idempotent)."""
cred = credentials(config)
delete(s3_address=key, cred=cred)
# ---------------------------------------------------------------------------
# exists
# ---------------------------------------------------------------------------
@cli.command("exists")
@_config_option
@click.option("--key", required=True, help="Object address — 's3://bucket/key' or bare key.")
def exists_(config: str | None, key: str) -> None:
"""Return exit code 0 if the object exists, 1 otherwise."""
cred = credentials(config)
present = exists(s3_address=key, cred=cred)
click.echo("true" if present else "false")
if not present:
sys.exit(1)
# ---------------------------------------------------------------------------
# list
# ---------------------------------------------------------------------------
@cli.command("list")
@_config_option
@click.option("--prefix", required=True, help="Key prefix (e.g. 'uploads/').")
@click.option("--max-keys", type=int, default=1000, show_default=True, help="Cap on returned keys.")
def list_cmd(config: str | None, prefix: str, max_keys: int) -> None:
"""List keys under a prefix in the default bucket."""
cred = credentials(config)
keys, truncated = list_prefix_page(prefix=prefix, cred=cred, max_keys=max_keys)
for key in keys:
click.echo(key)
# Keys stay on stdout so pipelines are unaffected; the caveat goes to
# stderr. Printing nothing here would let a capped listing read as a
# complete one, which is how an `xargs` over it quietly misses objects.
if truncated:
click.echo(
f"warning: more than {max_keys} keys under {prefix!r}; "
"raise --max-keys to see the rest.",
err=True,
)
# ---------------------------------------------------------------------------
# make-bucket
# ---------------------------------------------------------------------------
@cli.command("make-bucket")
@_config_option
@click.option("--bucket", required=True, help="Bucket name to create.")
def make_bucket_cmd(config: str | None, bucket: str) -> None:
"""Create a bucket (no-op if it already exists)."""
cred = credentials(config)
make_bucket(bucket=bucket, cred=cred)
# ---------------------------------------------------------------------------
# tempfile
# ---------------------------------------------------------------------------
@cli.command("tempfile")
@_config_option
@click.option("--ext", default=None, help="File extension for the generated name.")
@click.option("--prefix", default=None, help="Extra prefix path under the bucket / default prefix.")
def tempfile_(config: str | None, ext: str | None, prefix: str | None) -> None:
"""Emit a unique random key + public URL (JSON). Does NOT upload anything."""
cred = credentials(config)
with remote_tempfile(cred, ext=ext or "", prefix=prefix or "") as (addr, url):
click.echo(json.dumps({"s3_address": addr, "public_url": url}, indent=2))
# ---------------------------------------------------------------------------
# strip-path
# ---------------------------------------------------------------------------
@cli.command("strip-path")
@_config_option
@click.option("--address", required=True, help="Full 's3://bucket/key' URI or a bare key.")
def strip_path_cmd(config: str | None, address: str) -> None:
"""Extract the key part of an 's3://bucket/key' address."""
cred = credentials(config)
click.echo(strip_s3_path(address, cred))
[docs]
def main() -> None:
"""Console entry point (``bucket-helper``).
Click's own ``main()`` only special-cases ``ClickException``/``Abort``
(and a broken pipe); a plain library exception (a boto3 ``ClientError``,
the library's own ``RuntimeError``/``ValueError``, ...) would otherwise
propagate as a raw Python traceback instead of a clean CLI error. This
wraps the whole invocation and translates that last case into a one-line
stderr message + exit 1 — click's own control flow (usage errors,
``--help``, an explicit ``sys.exit(1)`` in a subcommand) already raises
``SystemExit``, a ``BaseException`` this does not catch, so it passes
through untouched.
"""
try:
cli()
except Exception as err: # noqa: BLE001 — last resort: see docstring
click.echo(f"Error: {err}", err=True)
sys.exit(1)
if __name__ == "__main__": # pragma: no cover
main()