Get a file from S3 (depends on boto3)

s3get.py

#!/usr/bin/env python3

import argparse
import logging

import boto3

logger = logging.getLogger(__name__)


def get_s3_resource(profile_name: str):
    if not profile_name:
        return boto3.resource("s3")

    session = boto3.Session(profile_name=profile_name)
    return session.resource("s3")


def parse_s3_uri(s3_uri: str) -> tuple[str, str]:
    if not s3_uri.startswith("s3://"):
        raise ValueError(f"Invalid S3 URI: {s3_uri}")

    uri = s3_uri[len("s3://"):]
    bucket_name, key = uri.split("/", 1)
    return bucket_name, key


def main(args: argparse.Namespace):
    s3 = get_s3_resource(args.profile)

    bucket_name, key = parse_s3_uri(args.s3_uri)
    bucket = s3.Bucket(bucket_name)

    if args.output_file:
        output_file = args.output_file
    else:
        output_file = key.split("/")[-1]

    logger.info("Downloading %s from %s to %s", key, bucket_name, output_file)
    bucket.download_file(key, output_file)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("s3_uri", help="s3://bucket/key")
    parser.add_argument("-o", "--output-file", help="Output file path")
    parser.add_argument("-p", "--profile", help="AWS profile name")
    return parser.parse_args()


if __name__ == "__main__":
    args = parse_args()
    logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
    main(args)