cli_app.py

"""
cli_app.py

Command-line entrypoint for the S3 file upload application.

Usage:
    python -m cli_app --source <local-file> --bucket <bucket>
                      [--key <object-key>] [--key-prefix <prefix>]
                      [--region <aws-region>] [--profile <aws-profile>]
                      [--content-type <mime>]
"""

import argparse
import sys
from typing import NoReturn

import local_file_helper
import s3_gateway
import transfer_models
import transfer_service

__all__ = ["main"]

# ---------------------------------------------------------------------------
# Private exit-code constants
# ---------------------------------------------------------------------------
_EXIT_SUCCESS = 0
_EXIT_RUNTIME_ERROR = 1
_EXIT_USAGE_ERROR = 2


# ---------------------------------------------------------------------------
# Private exception for CLI usage errors
# ---------------------------------------------------------------------------
class _CliUsageError(Exception):
    """Raised when the user provides invalid or missing CLI arguments."""


# ---------------------------------------------------------------------------
# Private ArgumentParser subclass that raises instead of calling sys.exit
# ---------------------------------------------------------------------------
class _ArgumentParser(argparse.ArgumentParser):
    """Custom parser that raises _CliUsageError instead of calling sys.exit."""

    def error(self, message: str) -> NoReturn:
        """Override to raise _CliUsageError instead of terminating the process."""
        raise _CliUsageError(message)


# ---------------------------------------------------------------------------
# Private helper: build the argument parser
# ---------------------------------------------------------------------------
def _build_parser() -> argparse.ArgumentParser:
    """Create and return the configured CLI argument parser."""
    parser = _ArgumentParser(
        prog="python -m cli_app",
        description="Upload a local file to an Amazon S3 bucket.",
    )
    parser.add_argument(
        "--source",
        required=True,
        metavar="LOCAL_FILE",
        help="Path to the local file to upload.",
    )
    parser.add_argument(
        "--bucket",
        required=True,
        metavar="BUCKET",
        help="Name of the target S3 bucket.",
    )
    parser.add_argument(
        "--key",
        required=False,
        default=None,
        metavar="OBJECT_KEY",
        help="Explicit S3 object key. If omitted, the filename is used.",
    )
    parser.add_argument(
        "--key-prefix",
        required=False,
        default=None,
        metavar="PREFIX",
        dest="key_prefix",
        help="Key prefix to prepend to the resolved object key.",
    )
    parser.add_argument(
        "--region",
        required=False,
        default=None,
        metavar="AWS_REGION",
        help="AWS region for the S3 client (e.g. us-east-1).",
    )
    parser.add_argument(
        "--profile",
        required=False,
        default=None,
        metavar="AWS_PROFILE",
        help="AWS CLI/boto3 named profile to use for credentials.",
    )
    parser.add_argument(
        "--content-type",
        required=False,
        default=None,
        metavar="MIME_TYPE",
        dest="content_type",
        help="MIME content type for the uploaded object (e.g. text/plain).",
    )
    return parser


# ---------------------------------------------------------------------------
# Private helper: convert parsed namespace to UploadRequest
# ---------------------------------------------------------------------------
def _namespace_to_request(args: argparse.Namespace) -> transfer_models.UploadRequest:
    """Convert parsed CLI arguments into an UploadRequest data object."""
    return transfer_models.UploadRequest(
        source_path=args.source,
        bucket=args.bucket,
        key=args.key,
        key_prefix=args.key_prefix,
        region=args.region,
        profile=args.profile,
        content_type=args.content_type,
    )


# ---------------------------------------------------------------------------
# Public entrypoint
# ---------------------------------------------------------------------------
def main(argv: list[str] | None = None) -> int:
    """
    Parse CLI arguments, wire dependencies, execute the S3 upload, and
    print the result.

    Parameters
    ----------
    argv:
        Argument vector to parse. When ``None``, ``sys.argv[1:]`` is used.

    Returns
    -------
    int
        Exit code: 0 on success, 1 on runtime/transfer error, 2 on usage error.
    """
    effective_argv: list[str] = sys.argv[1:] if argv is None else argv

    parser = _build_parser()

    # --- Parse arguments ---------------------------------------------------
    try:
        args = parser.parse_args(effective_argv)
    except _CliUsageError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return _EXIT_USAGE_ERROR
    except SystemExit as exc:
        # argparse raises SystemExit(0) after printing --help
        code = exc.code
        if code is None or code == 0:
            return _EXIT_SUCCESS
        return int(code)

    # --- Build request and execute transfer --------------------------------
    try:
        request = _namespace_to_request(args)

        inspector = local_file_helper.LocalFileInspector()
        uploader = s3_gateway.S3Uploader(region=request.region, profile=request.profile)
        service = transfer_service.FileTransferService(
            file_inspector=inspector,
            s3_uploader=uploader,
        )

        result: transfer_models.UploadResult = service.transfer(request)

    except Exception as exc:  # noqa: BLE001
        print(f"Error: {exc}", file=sys.stderr)
        return _EXIT_RUNTIME_ERROR

    # --- Report success ----------------------------------------------------
    print(f"s3://{result.bucket}/{result.key}")
    return _EXIT_SUCCESS


# ---------------------------------------------------------------------------
# Module execution hook
# ---------------------------------------------------------------------------
if __name__ == "__main__":
    raise SystemExit(main())