AWS S3 Multipart Upload with Progress

Problem Statement

Uploading large files (such as database dumps, machine learning datasets, or compressed archives) to Amazon S3 using single-part uploads is inherently risky. Standard uploads are prone to network timeouts, connection resets, and strict 5 GB object limits. If a 40 GB file fails at 99% completion during a single-part upload, the entire transfer must be restarted from the beginning, consuming unnecessary bandwidth and time.

This script solves the problem by leveraging S3 Multipart Uploads configured with parallel chunk transfers and custom progress tracking. Large files are split into managed parts, uploaded concurrently, and reassembled in S3 upon completion. If a single part fails, only that specific chunk is retried. Additionally, the script provides real-time progress metrics (transferred bytes, speed in MB/s, percentage, and estimated time remaining) without relying on external UI dependencies like tqdm.

How It Works

The script utilizes boto3's high-level S3 Transfer Manager via s3_client.upload_file(). This API automatically abstracts low-level multipart operations (CreateMultipartUpload, UploadPart, CompleteMultipartUpload) while offering fine-grained performance tuning using boto3.s3.transfer.TransferConfig.

Key components and design patterns include:

• TransferConfig Tuning: Converts files larger than a specified threshold (default: 8 MB) into multipart transfers, streaming chunks (default: 8 MB per part) across up to 10 concurrent worker threads.
• Thread-Safe Callback Pattern: Boto3 invokes a callback function on thread pools whenever bytes are transmitted. A thread-safe ProgressPercentage class uses threading.Lock to safely accumulate transferred byte counts across parallel workers.
• Dynamic Terminal UI: Calculates throughput dynamically using elapsed time and formats terminal output in real time using ANSI carriage returns (\r) to prevent terminal log spam.

+-----------------------------------------------------------------------+
|                             Local System                              |
|  +------------------+     Split Chunks    +-------------------------+ |
|  |  Large File      |  ------------->     | Thread Pool (Worker 1)  | |
|  | (e.g., 10 GB)    |                     | Thread Pool (Worker 2)  | |
|  +------------------+                     | Thread Pool (Worker N)  | |
+-----------------------------------------------+-----------------------+
                                                |  Parallel HTTP PUTs
                                                v
                                  +---------------------------+
                                  |     AWS S3 Bucket         |
                                  | (Reassembled Automatically)|
                                  +---------------------------+

Full Script

#!/usr/bin/env python3
"""
AWS S3 Multipart Upload with Progress

Performs concurrent multipart uploads to AWS S3 with real-time progress,
speed estimation, and customizable chunk configurations.
"""

import argparse
import os
import sys
import time
import threading
from pathlib import Path
from typing import Optional

import boto3
from boto3.s3.transfer import TransferConfig
from botocore.exceptions import BotoCoreError, ClientError


class ProgressPercentage:
    """Thread-safe progress callback for Boto3 S3 transfers."""

    def __init__(self, filename: str, total_bytes: int) -> None:
        self._filename: str = filename
        self._total_bytes: int = total_bytes
        self._seen_so_far: int = 0
        self._lock: threading.Lock = threading.Lock()
        self._start_time: float = time.time()

    def __call__(self, bytes_amount: int) -> None:
        with self._lock:
            self._seen_so_far += bytes_amount
            elapsed: float = time.time() - self._start_time
            
            # Avoid divide-by-zero on ultra-fast transfers or tiny initial frames
            speed_mbps: float = (self._seen_so_far / (1024 * 1024)) / elapsed if elapsed > 0 else 0.0
            percentage: float = (self._seen_so_far / self._total_bytes) * 100

            remaining_bytes: int = self._total_bytes - self._seen_so_far
            eta_seconds: float = (remaining_bytes / (1024 * 1024)) / speed_mbps if speed_mbps > 0 else 0

            # Format byte output (MB or GB)
            seen_mb: float = self._seen_so_far / (1024 * 1024)
            total_mb: float = self._total_bytes / (1024 * 1024)

            sys.stdout.write(
                f"\rUploading {Path(self._filename).name}: "
                f"[{percentage:6.2f}%] {seen_mb:.1f}/{total_mb:.1f} MB "
                f"@{speed_mbps:6.2f} MB/s (ETA: {int(eta_seconds)}s)"
            )
            sys.stdout.flush()


def upload_file_to_s3(
    file_path: str,
    bucket: str,
    object_name: Optional[str] = None,
    region: Optional[str] = None,
    chunk_size_mb: int = 8,
    max_concurrency: int = 10,
) -> bool:
    """Uploads a local file to S3 using multipart configuration and progress reporting.

    Args:
        file_path: Path to the local file to upload.
        bucket: Name of the destination S3 bucket.
        object_name: S3 key name. Defaults to the base filename.
        region: AWS region. Defaults to system/env configuration.
        chunk_size_mb: Size of each multipart chunk in megabytes.
        max_concurrency: Number of parallel upload threads.

    Returns:
        True if upload succeeded, False otherwise.
    """
    path = Path(file_path)
    if not path.is_file():
        print(f"Error: Local file '{file_path}' does not exist.", file=sys.stderr)
        return False

    total_bytes = path.stat().st_size
    s3_key = object_name if object_name else path.name

    # Initialize S3 client with optional region override
    client_kwargs = {}
    if region:
        client_kwargs["region_name"] = region

    s3_client = boto3.client("s3", **client_kwargs)

    # Convert MB settings to bytes for TransferConfig
    chunk_bytes = chunk_size_mb * 1024 * 1024
    
    # Configure multipart upload thresholds
    config = TransferConfig(
        multipart_threshold=chunk_bytes,
        max_concurrency=max_concurrency,
        multipart_chunksize=chunk_bytes,
        use_threads=True,
    )

    print(f"Target Bucket : {bucket}")
    print(f"Target Key    : {s3_key}")
    print(f"File Size     : {total_bytes / (1024 * 1024):.2f} MB")
    print(f"Chunk Size    : {chunk_size_mb} MB (Concurrency: {max_concurrency})")
    print("-" * 60)

    progress = ProgressPercentage(file_path, total_bytes)

    try:
        s3_client.upload_file(
            Filename=str(path),
            Bucket=bucket,
            Key=s3_key,
            Callback=progress,
            Config=config,
        )
        print("\nUpload completed successfully!")
        return True

    except FileNotFoundError:
        print(f"\nError: File not found during execution: {file_path}", file=sys.stderr)
        return False
    except ClientError as e:
        error_code = e.response.get("Error", {}).get("Code", "Unknown")
        print(f"\nAWS S3 Client Error [{error_code}]: {e}", file=sys.stderr)
        return False
    except BotoCoreError as e:
        print(f"\nAWS BotoCore Error: {e}", file=sys.stderr)
        return False
    except KeyboardInterrupt:
        print("\nUpload cancelled by user.", file=sys.stderr)
        return False


def parse_args() -> argparse.Namespace:
    """Parses command-line arguments and merges with environment defaults."""
    parser = argparse.ArgumentParser(
        description="Upload large files to AWS S3 with multipart configuration and progress reporting."
    )
    
    parser.add_argument(
        "-f", "--file",
        required=True,
        help="Path to the local file to upload"
    )
    parser.add_argument(
        "-b", "--bucket",
        default=os.environ.get("AWS_S3_BUCKET"),
        help="Target S3 bucket name (or set AWS_S3_BUCKET env var)"
    )
    parser.add_argument(
        "-k", "--key",
        default=None,
        help="Target S3 key path (defaults to local filename)"
    )
    parser.add_argument(
        "-r", "--region",
        default=os.environ.get("AWS_REGION"),
        help="AWS region (or set AWS_REGION env var)"
    )
    parser.add_argument(
        "--chunk-size-mb",
        type=int,
        default=8,
        help="Multipart chunk size in MB (default: 8 MB, min: 5 MB)"
    )
    parser.add_argument(
        "--concurrency",
        type=int,
        default=10,
        help="Maximum concurrent upload threads (default: 10)"
    )

    args = parser.parse_args()

    if not args.bucket:
        parser.error("Bucket name is required. Specify via --bucket or AWS_S3_BUCKET environment variable.")

    if args.chunk_size_mb < 5:
        parser.error("S3 requires multipart chunks to be at least 5 MB.")

    return args


if __name__ == "__main__":
    cli_args = parse_args()

    success = upload_file_to_s3(
        file_path=cli_args.file,
        bucket=cli_args.bucket,
        object_name=cli_args.key,
        region=cli_args.region,
        chunk_size_mb=cli_args.chunk_size_mb,
        max_concurrency=cli_args.concurrency,
    )

    sys.exit(0 if success else 1)

Configuration

The script accepts settings via standard AWS environment variables, custom environment variables, and CLI arguments. CLI flags take precedence over environment settings.

Usage Examples

Standard Upload using CLI Arguments

Upload a large compressed archive directly to an S3 bucket:

python3 s3_upload.py --file large_dataset.tar.gz --bucket my-company-backups

Output:

Target Bucket : my-company-backups
Target Key    : large_dataset.tar.gz
File Size     : 1250.40 MB
Chunk Size    : 8 MB (Concurrency: 10)
------------------------------------------------------------
Uploading large_dataset.tar.gz: [ 45.30%] 566.4/1250.4 MB @ 24.50 MB/s (ETA: 27s)

Advanced Usage with Custom Prefix and Tuning

Specify an S3 object key destination along with custom chunk size and thread concurrency for high-bandwidth networks:

python3 s3_upload.py \
  --file /var/backups/db_dump.sql \
  --bucket production-db-backups \
  --key postgres/2026-03-30/db_dump.sql \
  --chunk-size-mb 16 \
  --concurrency 16

Environment Variable Integration

Define target bucket and region inside environment variables to simplify repetitive commands:

export AWS_S3_BUCKET="my-automation-bucket"
export AWS_REGION="us-east-1"

python3 s3_upload.py --file ML_Model_v2.bin

Error Handling

The script handles common network, authorization, and execution errors gracefully:

1. Missing Local Files: Checks file presence using pathlib.Path.is_file() before initiating S3 connections, outputting a descriptive error message.
2. Minimum Chunk Size Violation: Enforces AWS S3's strict 5 MB minimum limit on multipart parts (--chunk-size-mb < 5), throwing an early validation error.
3. Invalid AWS Credentials: Catches botocore.exceptions.ClientError (e.g., InvalidAccessKeyId, SignatureDoesNotMatch, AccessDenied) and displays human-readable error codes.
4. Bucket Non-Existence: Traps 404 NoSuchBucket errors when targeting missing or misspelled S3 buckets.
5. Thread Safety & SIGINT Handling: The progress class utilizes threading.Lock() to prevent race conditions during updates from concurrent worker threads. Keyboard interrupts (Ctrl+C) are caught safely without printing traceback spam.

Extending Ideas

• Server-Side Encryption (SSE): Pass ExtraArgs={'ServerSideEncryption': 'aws:kms', 'SSEKMSKeyId': 'key-id'} into s3_client.upload_file() for enforced encryption compliance.
• Automatic MD5/SHA256 Checksum Verification: Calculate local file checksums prior to upload and attach them using S3's ChecksumAlgorithm parameters (SHA256 or CRC32).
• Storage Class Selection: Add a CLI flag (--storage-class) allowing users to target INTELLIGENT_TIERING, STANDARD_IA, or GLACIER_IR storage classes directly upon upload.
• Webhook Notifications: Integrate a completion hook using standard libraries (e.g., urllib.request) to post upload metrics to a Discord or Slack webhook when large nightly automated backups finish.