AWS S3 Helper Script

env.example

AWS_REGION=us-west-1
AWS_ACCESS_KEY_ID=your-access-key-id
AWS_ACCESS_KEY_SECRET=your-secret-access-key
S3_PATH=s3://your-bucket/optional/prefix

pyproject.toml

[project]
name = "helpers-s3"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
    "boto3",
    "python-dotenv",
]

[project.scripts]
s3helper = "s3helper:main"

[tool.uv]
package = true

s3-helpers.bat

@ECHO OFF
SETLOCAL ENABLEDELAYEDEXPANSION

REM ── Load .env ────────────────────────────────────────────────────────────────
SET "_SCRIPT_DIR=%~dp0"
IF EXIST "%_SCRIPT_DIR%.env" (
    FOR /F "usebackq tokens=1,* delims==" %%A IN ("%_SCRIPT_DIR%.env") DO (
        SET "_LINE=%%A"
        IF NOT "!_LINE:~0,1!"=="#" IF NOT "!_LINE!"=="" (
            SET "%%A=%%B"
        )
    )
)

REM ── Initialise argument variables ─────────────────────────────────────────────
SET "_CMD="
SET "_FILE="
SET "_S3PATH="
SET "_LIST_INLINE="

REM ── Parse arguments ───────────────────────────────────────────────────────────
:parse
IF "%~1"=="" GOTO validate
IF /I "%~1"=="--upload"  ( SET "_CMD=upload"  & SHIFT & GOTO parse )
IF /I "%~1"=="--list"    GOTO handle_list
IF /I "%~1"=="--delete"  ( SET "_CMD=delete"  & SHIFT & GOTO parse )
IF /I "%~1"=="--empty"   ( SET "_CMD=empty"   & SHIFT & GOTO parse )
IF /I "%~1"=="--file"    ( SET "_FILE=%~2"    & SHIFT & SHIFT & GOTO parse )
IF /I "%~1"=="--s3Path"  ( SET "_S3PATH=%~2"  & SHIFT & SHIFT & GOTO parse )
ECHO Unknown argument: %~1
EXIT /B 1

REM --list can take an optional inline s3:// path as the very next token.
REM We must handle this via a label (not a parenthesized block) so that %~1
REM is re-evaluated AFTER the SHIFT, not captured at block-parse time.
:handle_list
SET "_CMD=list"
SHIFT
SET "_NEXT=%~1"
IF NOT "%_NEXT%"=="" (
    IF NOT "!_NEXT:~0,2!"=="--" (
        SET "_LIST_INLINE=%_NEXT%"
        SHIFT
    )
)
GOTO parse

:validate
IF "%_CMD%"=="" (
    ECHO Usage: s3helper.bat --upload ^| --list [S3_PATH] ^| --delete ^| --empty
    ECHO        [--file PATH] [--s3Path PATH]
    EXIT /B 1
)

REM ── Validate required env vars ────────────────────────────────────────────────
IF "%AWS_REGION%"==""           ( ECHO Missing required environment variable: AWS_REGION            & EXIT /B 1 )
IF "%AWS_ACCESS_KEY_ID%"==""    ( ECHO Missing required environment variable: AWS_ACCESS_KEY_ID     & EXIT /B 1 )
IF "%AWS_ACCESS_KEY_SECRET%"="" ( ECHO Missing required environment variable: AWS_ACCESS_KEY_SECRET & EXIT /B 1 )
IF "%S3_PATH%"==""              ( ECHO Missing required environment variable: S3_PATH               & EXIT /B 1 )

REM ── Map to the env var names the AWS CLI expects ──────────────────────────────
SET "AWS_SECRET_ACCESS_KEY=%AWS_ACCESS_KEY_SECRET%"
SET "AWS_DEFAULT_REGION=%AWS_REGION%"

REM ── Dispatch ──────────────────────────────────────────────────────────────────
IF "%_CMD%"=="upload" GOTO cmd_upload
IF "%_CMD%"=="list"   GOTO cmd_list
IF "%_CMD%"=="delete" GOTO cmd_delete
IF "%_CMD%"=="empty"  GOTO cmd_empty
EXIT /B 1

REM ─────────────────────────────────────────────────────────────────────────────
:cmd_upload
IF "%_FILE%"=="" ( ECHO --upload requires --file ^<local_path^> & EXIT /B 1 )

REM Extract just the filename (strip directory component)
FOR %%F IN ("%_FILE%") DO SET "_FNAME=%%~nxF"

REM Resolve destination URI — mirrors resolve_upload_destination() in s3helper.py
IF "%_S3PATH%"=="" (
    CALL :strip_trailing_slash "%S3_PATH%" _BASE
    SET "_DEST=!_BASE!/!_FNAME!"
) ELSE (
    CALL :starts_with_s3 "%_S3PATH%" _IS_ABS
    IF "!_IS_ABS!"=="1" (
        SET "_LAST=!_S3PATH:~-1!"
        IF "!_LAST!"=="/" (
            SET "_DEST=!_S3PATH!!_FNAME!"
        ) ELSE (
            SET "_DEST=!_S3PATH!"
        )
    ) ELSE (
        CALL :strip_trailing_slash "%S3_PATH%" _BASE
        CALL :strip_trailing_slash "%_S3PATH%" _PREFIX
        SET "_DEST=!_BASE!/!_PREFIX!/!_FNAME!"
    )
)

aws s3 cp "%_FILE%" "!_DEST!"
IF ERRORLEVEL 1 EXIT /B 1
ECHO Uploaded: !_DEST!
EXIT /B 0

REM ─────────────────────────────────────────────────────────────────────────────
:cmd_list
IF DEFINED _LIST_INLINE (
    SET "_PREFIX_URI=!_LIST_INLINE!"
) ELSE IF NOT "%_S3PATH%"=="" (
    SET "_PREFIX_URI=%_S3PATH%"
) ELSE (
    SET "_PREFIX_URI=%S3_PATH%"
)

aws s3 ls "!_PREFIX_URI!" --recursive
IF ERRORLEVEL 1 EXIT /B 1
EXIT /B 0

REM ─────────────────────────────────────────────────────────────────────────────
:cmd_delete
IF "%_FILE%"=="" ( ECHO --delete requires --file ^<s3_path_or_relative^> & EXIT /B 1 )

CALL :starts_with_s3 "%_FILE%" _IS_ABS
IF "!_IS_ABS!"=="1" (
    SET "_TARGET=%_FILE%"
) ELSE (
    CALL :strip_trailing_slash "%S3_PATH%" _BASE
    CALL :strip_leading_slash "%_FILE%" _REL
    SET "_TARGET=!_BASE!/!_REL!"
)

REM Check existence via head-object (mirrors Python's client.head_object call)
CALL :parse_s3_uri "!_TARGET!" _BUCKET _KEY
aws s3api head-object --bucket "!_BUCKET!" --key "!_KEY!" >NUL 2>&1
IF ERRORLEVEL 1 (
    ECHO Not found: !_TARGET!
    EXIT /B 0
)
aws s3 rm "!_TARGET!"
IF ERRORLEVEL 1 EXIT /B 1
ECHO Deleted: !_TARGET!
EXIT /B 0

REM ─────────────────────────────────────────────────────────────────────────────
:cmd_empty
SET /P "_ANSWER=Empty all objects under %S3_PATH% ? [y/N] "
IF /I NOT "!_ANSWER!"=="y" ( ECHO Aborted. & EXIT /B 0 )
aws s3 rm "%S3_PATH%" --recursive
IF ERRORLEVEL 1 EXIT /B 1
EXIT /B 0

REM ─────────────────────────────────────────────────────────────────────────────
REM Subroutine: strip trailing / or \ from %~1, store result in var named %~2
:strip_trailing_slash
SET "_STS=%~1"
IF "!_STS:~-1!"=="/" SET "_STS=!_STS:~0,-1!"
IF "!_STS:~-1!"=="\" SET "_STS=!_STS:~0,-1!"
SET "%~2=!_STS!"
EXIT /B 0

REM Subroutine: strip leading / or \ from %~1, store result in var named %~2
:strip_leading_slash
SET "_SLS=%~1"
IF "!_SLS:~0,1!"=="/" SET "_SLS=!_SLS:~1!"
IF "!_SLS:~0,1!"=="\" SET "_SLS=!_SLS:~1!"
SET "%~2=!_SLS!"
EXIT /B 0

REM Subroutine: set %~2=1 if %~1 starts with s3://, else 0
:starts_with_s3
SET "_SWS=%~1"
SET "%~2=0"
IF "!_SWS:~0,5!"=="s3://" SET "%~2=1"
EXIT /B 0

REM Subroutine: split an s3://bucket/key URI into bucket (%~2) and key (%~3)
:parse_s3_uri
SET "_PSU=%~1"
SET "_PSU=!_PSU:s3://=!"
FOR /F "tokens=1,* delims=/" %%B IN ("!_PSU!") DO (
    SET "%~2=%%B"
    SET "%~3=%%C"
)
EXIT /B 0

S3-Helpers.md

helpers-s3

A command-line tool for common AWS S3 operations: upload files, list objects, delete individual objects, and empty an entire S3 prefix.

Prerequisites

• Python >= 3.11
• uv package manager
• AWS IAM credentials with S3 read/write access

Setup

1. Install dependencies

uv sync

2. Configure environment

cp env.example .env

Edit .env with your values:

Variable	Description	Example
AWS_REGION	AWS region	us-east-1
AWS_ACCESS_KEY_ID	IAM access key ID	AKIA...
AWS_ACCESS_KEY_SECRET	IAM secret access key	abc123...
S3_PATH	Default S3 bucket + prefix	s3://my-bucket/prefix

Usage

All commands are mutually exclusive — use one per invocation.

List objects

# Uses S3_PATH from .env
s3helper --list

# Override with a specific path
s3helper --list s3://my-bucket/some/prefix

Upload a file

# Upload to S3_PATH/<filename>
s3helper --upload --file ./local-file.csv

# Upload to a relative destination (resolved under S3_PATH bucket)
s3helper --upload --file ./local-file.csv --s3Path data/file.csv

# Upload to a full S3 URI
s3helper --upload --file ./local-file.csv --s3Path s3://my-bucket/data/file.csv

Delete an object

# Relative path (resolved against S3_PATH)
s3helper --delete --file relative/path/file.csv

# Full S3 URI
s3helper --delete --file s3://my-bucket/path/file.csv

Empty all objects under S3_PATH

s3helper --empty

Prompts for confirmation before deleting anything.

Help

s3helper --help

s3helper.py

import argparse
import os
import sys
from pathlib import Path

import boto3
from botocore.exceptions import ClientError
from dotenv import load_dotenv

load_dotenv()


def _require_env(name: str) -> str:
    val = os.getenv(name)
    if not val:
        sys.exit(f"Missing required environment variable: {name}")
    return val


def _s3_client():
    return boto3.client(
        "s3",
        region_name=_require_env("AWS_REGION"),
        aws_access_key_id=_require_env("AWS_ACCESS_KEY_ID"),
        aws_secret_access_key=_require_env("AWS_ACCESS_KEY_SECRET"),
    )


def parse_s3_uri(uri: str) -> tuple[str, str]:
    """Split s3://bucket/key into (bucket, key)."""
    without_scheme = uri[len("s3://"):]
    bucket, _, key = without_scheme.partition("/")
    return bucket, key


def resolve_s3_path(path: str, default_base: str) -> str:
    """Return a full s3:// URI. If path already starts with s3://, use it directly."""
    if path.startswith("s3://"):
        return path
    base = default_base.rstrip("/")
    return f"{base}/{path.lstrip('/')}"


def resolve_upload_destination(file_path: str, s3_path: str | None, default_base: str) -> str:
    """Determine the full s3:// destination URI for an upload."""
    filename = Path(file_path).name
    if s3_path is None:
        return f"{default_base.rstrip('/')}/{filename}"
    if s3_path.startswith("s3://"):
        # explicit full URI — use as-is (treat as object key if it ends with /)
        if s3_path.endswith("/"):
            return f"{s3_path}{filename}"
        return s3_path
    # relative path: place under default_base
    base = default_base.rstrip("/")
    prefix = s3_path.rstrip("/")
    return f"{base}/{prefix}/{filename}"


def cmd_upload(args):
    if not args.file:
        sys.exit("--upload requires --file <local_path>")
    default_base = _require_env("S3_PATH")
    destination = resolve_upload_destination(args.file, args.s3Path, default_base)
    bucket, key = parse_s3_uri(destination)
    client = _s3_client()
    client.upload_file(args.file, bucket, key)
    print(f"Uploaded: s3://{bucket}/{key}")


def cmd_list(args):
    default_base = _require_env("S3_PATH")
    inline = args.list if isinstance(args.list, str) else None
    prefix_uri = inline or args.s3Path or default_base
    bucket, prefix = parse_s3_uri(prefix_uri)
    client = _s3_client()
    paginator = client.get_paginator("list_objects_v2")
    found = False
    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        for obj in page.get("Contents", []):
            print(f"s3://{bucket}/{obj['Key']}")
            found = True
    if not found:
        print(f"No objects found under s3://{bucket}/{prefix}")


def cmd_delete(args):
    if not args.file:
        sys.exit("--delete requires --file <s3_path_or_relative>")
    default_base = _require_env("S3_PATH")
    target_uri = resolve_s3_path(args.file, default_base)
    bucket, key = parse_s3_uri(target_uri)
    client = _s3_client()
    try:
        client.head_object(Bucket=bucket, Key=key)
    except ClientError as exc:
        if exc.response["Error"]["Code"] in ("404", "NoSuchKey"):
            print(f"Not found: {target_uri}")
            return
        raise
    client.delete_object(Bucket=bucket, Key=key)
    print(f"Deleted: {target_uri}")


def cmd_empty(args):
    default_base = _require_env("S3_PATH")
    bucket, prefix = parse_s3_uri(default_base)
    answer = input(f"Empty all objects under s3://{bucket}/{prefix} ? [y/N] ").strip().lower()
    if answer != "y":
        print("Aborted.")
        return
    client = _s3_client()
    paginator = client.get_paginator("list_objects_v2")
    deleted = 0
    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        objects = [{"Key": obj["Key"]} for obj in page.get("Contents", [])]
        if objects:
            client.delete_objects(Bucket=bucket, Delete={"Objects": objects})
            deleted += len(objects)
    print(f"Deleted {deleted} object(s).")


def main():
    parser = argparse.ArgumentParser(description="Simple S3 helper CLI")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--upload", action="store_true", help="Upload a file to S3")
    group.add_argument("--list", nargs="?", const=True, metavar="S3_PATH",
                       help="List objects in S3; optionally pass an s3:// path to override S3_PATH")
    group.add_argument("--delete", action="store_true", help="Delete an object from S3")
    group.add_argument("--empty", action="store_true", help="Delete all objects under S3_PATH")
    parser.add_argument("--file", help="Local file path (upload) or S3 key/URI (delete)")
    parser.add_argument("--s3Path", help="Destination S3 path or prefix override")

    args = parser.parse_args()

    if args.upload:
        cmd_upload(args)
    elif args.list:
        cmd_list(args)
    elif args.delete:
        cmd_delete(args)
    elif args.empty:
        cmd_empty(args)


if __name__ == "__main__":
    main()

uv.lock

version = 1
revision = 3
requires-python = ">=3.11"

[[package]]
name = "boto3"
version = "1.42.93"
source = { registry = "https://pypi.org/simple" }
dependencies = [
    { name = "botocore" },
    { name = "jmespath" },
    { name = "s3transfer" },
]
sdist = { url = "https://files.pythonhosted.org/packages/dd/ac/e6b2b24d53c830500176f710594efcde626186b5b3c9aead6f8837976956/boto3-1.42.93.tar.gz", hash = "sha256:ff81c6bac708cb95c4f8b27e331ac67d95c6908dd86bcb7b15b8941960f2bc4c", size = 113218, upload-time = "2026-04-21T21:30:39.733Z" }
wheels = [
    { url = "https://files.pythonhosted.org/packages/f2/2d/fcc35bde9fa47ac463a3023c73838e23e9281cde7f5e86fe7c459c3b72aa/boto3-1.42.93-py3-none-any.whl", hash = "sha256:51e34e30e65bea4df0ff77f91abdcb97297eb74c3b27eb576b2abbd758452967", size = 140554, upload-time = "2026-04-21T21:30:36.581Z" },
]

[[package]]
name = "botocore"
version = "1.42.93"
source = { registry = "https://pypi.org/simple" }
dependencies = [
    { name = "jmespath" },
    { name = "python-dateutil" },
    { name = "urllib3" },
]
sdist = { url = "https://files.pythonhosted.org/packages/1b/d4/eb53f7ed81836696abf7103c9c901a0cace9217328094ca93419016a78c9/botocore-1.42.93.tar.gz", hash = "sha256:9ce49863c50b43f7942edd295fb16bfc6d227264ce6fc32c8f2426ef11b9351b", size = 15239759, upload-time = "2026-04-21T21:30:23.707Z" }
wheels = [
    { url = "https://files.pythonhosted.org/packages/f4/0c/ccc57c9a7bcd4553620bf6f50a3640ba68d189330fc4787dbddb2d851534/botocore-1.42.93-py3-none-any.whl", hash = "sha256:96ae26cd6302a7c7563398517b90a438168a4efdf4f73ab38882cefb8df721cc", size = 14923656, upload-time = "2026-04-21T21:30:17.597Z" },
]

[[package]]
name = "helpers-s3"
version = "0.1.0"
source = { editable = "." }
dependencies = [
    { name = "boto3" },
    { name = "python-dotenv" },
]

[package.metadata]
requires-dist = [
    { name = "boto3" },
    { name = "python-dotenv" },
]

[[package]]
name = "jmespath"
version = "1.1.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/d3/59/322338183ecda247fb5d1763a6cbe46eff7222eaeebafd9fa65d4bf5cb11/jmespath-1.1.0.tar.gz", hash = "sha256:472c87d80f36026ae83c6ddd0f1d05d4e510134ed462851fd5f754c8c3cbb88d", size = 27377, upload-time = "2026-01-22T16:35:26.279Z" }
wheels = [
    { url = "https://files.pythonhosted.org/packages/14/2f/967ba146e6d58cf6a652da73885f52fc68001525b4197effc174321d70b4/jmespath-1.1.0-py3-none-any.whl", hash = "sha256:a5663118de4908c91729bea0acadca56526eb2698e83de10cd116ae0f4e97c64", size = 20419, upload-time = "2026-01-22T16:35:24.919Z" },
]

[[package]]
name = "python-dateutil"
version = "2.9.0.post0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
    { name = "six" },
]
sdist = { url = "https://files.pythonhosted.org/packages/66/c0/0c8b6ad9f17a802ee498c46e004a0eb49bc148f2fd230864601a86dcf6db/python-dateutil-2.9.0.post0.tar.gz", hash = "sha256:37dd54208da7e1cd875388217d5e00ebd4179249f90fb72437e91a35459a0ad3", size = 342432, upload-time = "2024-03-01T18:36:20.211Z" }
wheels = [
    { url = "https://files.pythonhosted.org/packages/ec/57/56b9bcc3c9c6a792fcbaf139543cee77261f3651ca9da0c93f5c1221264b/python_dateutil-2.9.0.post0-py2.py3-none-any.whl", hash = "sha256:a8b2bc7bffae282281c8140a97d3aa9c14da0b136dfe83f850eea9a5f7470427", size = 229892, upload-time = "2024-03-01T18:36:18.57Z" },
]

[[package]]
name = "python-dotenv"
version = "1.2.2"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/82/ed/0301aeeac3e5353ef3d94b6ec08bbcabd04a72018415dcb29e588514bba8/python_dotenv-1.2.2.tar.gz", hash = "sha256:2c371a91fbd7ba082c2c1dc1f8bf89ca22564a087c2c287cd9b662adde799cf3", size = 50135, upload-time = "2026-03-01T16:00:26.196Z" }
wheels = [
    { url = "https://files.pythonhosted.org/packages/0b/d7/1959b9648791274998a9c3526f6d0ec8fd2233e4d4acce81bbae76b44b2a/python_dotenv-1.2.2-py3-none-any.whl", hash = "sha256:1d8214789a24de455a8b8bd8ae6fe3c6b69a5e3d64aa8a8e5d68e694bbcb285a", size = 22101, upload-time = "2026-03-01T16:00:25.09Z" },
]

[[package]]
name = "s3transfer"
version = "0.16.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
    { name = "botocore" },
]
sdist = { url = "https://files.pythonhosted.org/packages/05/04/74127fc843314818edfa81b5540e26dd537353b123a4edc563109d8f17dd/s3transfer-0.16.0.tar.gz", hash = "sha256:8e990f13268025792229cd52fa10cb7163744bf56e719e0b9cb925ab79abf920", size = 153827, upload-time = "2025-12-01T02:30:59.114Z" }
wheels = [
    { url = "https://files.pythonhosted.org/packages/fc/51/727abb13f44c1fcf6d145979e1535a35794db0f6e450a0cb46aa24732fe2/s3transfer-0.16.0-py3-none-any.whl", hash = "sha256:18e25d66fed509e3868dc1572b3f427ff947dd2c56f844a5bf09481ad3f3b2fe", size = 86830, upload-time = "2025-12-01T02:30:57.729Z" },
]

[[package]]
name = "six"
version = "1.17.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/94/e7/b2c673351809dca68a0e064b6af791aa332cf192da575fd474ed7d6f16a2/six-1.17.0.tar.gz", hash = "sha256:ff70335d468e7eb6ec65b95b99d3a2836546063f63acc5171de367e834932a81", size = 34031, upload-time = "2024-12-04T17:35:28.174Z" }
wheels = [
    { url = "https://files.pythonhosted.org/packages/b7/ce/149a00dd41f10bc29e5921b496af8b574d8413afcd5e30dfa0ed46c2cc5e/six-1.17.0-py2.py3-none-any.whl", hash = "sha256:4721f391ed90541fddacab5acf947aa0d3dc7d27b2e1e8eda2be8970586c3274", size = 11050, upload-time = "2024-12-04T17:35:26.475Z" },
]

[[package]]
name = "urllib3"
version = "2.6.3"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/c7/24/5f1b3bdffd70275f6661c76461e25f024d5a38a46f04aaca912426a2b1d3/urllib3-2.6.3.tar.gz", hash = "sha256:1b62b6884944a57dbe321509ab94fd4d3b307075e0c2eae991ac71ee15ad38ed", size = 435556, upload-time = "2026-01-07T16:24:43.925Z" }
wheels = [
    { url = "https://files.pythonhosted.org/packages/39/08/aaaad47bc4e9dc8c725e68f9d04865dbcb2052843ff09c97b08904852d84/urllib3-2.6.3-py3-none-any.whl", hash = "sha256:bf272323e553dfb2e87d9bfd225ca7b0f467b919d7bbd355436d3fd37cb0acd4", size = 131584, upload-time = "2026-01-07T16:24:42.685Z" },
]