S3

import boto3

s3 = boto3.client("s3")

# Upload / download files
s3.upload_file("local.txt", "my-bucket", "remote.txt")
s3.download_file("my-bucket", "remote.txt", "local.txt")

# Upload / download file-like objects
with open("local.txt", "rb") as f:
    s3.upload_fileobj(f, "my-bucket", "remote.txt")

# Presigned URL
url = s3.generate_presigned_url(
    "get_object",
    Params={"Bucket": "my-bucket", "Key": "my-key"},
    ExpiresIn=3600,
)

response = s3.get_object(Bucket="bucket", Key="key")
try:
    data = response["Body"].read()
finally:
    response["Body"].close()

# Or use as context manager
with s3.get_object(Bucket="bucket", Key="key")["Body"] as body:
    data = body.read()

import boto3

s3 = boto3.client("s3")

# File transfers
s3.upload_file("local.txt", "my-bucket", "remote.txt")
s3.download_file("my-bucket", "remote.txt", "local.txt")

# File-like object transfers
with open("local.txt", "rb") as f:
    s3.upload_fileobj(f, "my-bucket", "remote.txt")

with open("local.txt", "wb") as f:
    s3.download_fileobj("my-bucket", "remote.txt", f)

s3.upload_file(
    "local.txt", "my-bucket", "remote.txt",
    ExtraArgs={
        "ContentType": "text/plain",
        "ServerSideEncryption": "aws:kms",
        "Metadata": {"author": "alice"},
    },
)

import os

file_size = os.path.getsize("large_file.bin")
uploaded = 0

def progress(bytes_transferred):
    nonlocal uploaded
    uploaded += bytes_transferred
    pct = (uploaded / file_size) * 100
    print(f"\r{pct:.1f}%", end="")

s3.upload_file("large_file.bin", "bucket", "key", Callback=progress)

from boto3.s3.transfer import TransferConfig

config = TransferConfig(
    multipart_threshold=8 * 1024 * 1024,   # switch to multipart above 8MB (default 8MB)
    max_concurrency=10,                     # parallel transfer threads (default 10)
    multipart_chunksize=8 * 1024 * 1024,   # part size (default 8MB, min 5MB)
    use_threads=True,                       # enable threading (default True)
)

s3.upload_file("large.bin", "bucket", "key", Config=config)
s3.download_file("bucket", "key", "large.bin", Config=config)

response = s3.get_object(Bucket="bucket", Key="key")
body = response["Body"]

# Read all at once
data = body.read()
body.close()

# Read in chunks
for chunk in body.iter_chunks(chunk_size=4096):
    process(chunk)
body.close()

# Read lines (for text content)
for line in body.iter_lines():
    process(line)

# As context manager -- auto-closes
with s3.get_object(Bucket="bucket", Key="key")["Body"] as body:
    data = body.read()

url = s3.generate_presigned_url(
    "get_object",
    Params={"Bucket": "bucket", "Key": "key"},
    ExpiresIn=3600,  # seconds (default 3600)
)

url = s3.generate_presigned_url(
    "put_object",
    Params={
        "Bucket": "bucket",
        "Key": "key",
        "ContentType": "application/pdf",
    },
    ExpiresIn=3600,
)
# Client must include Content-Type: application/pdf in the upload request

presigned = s3.generate_presigned_post(
    Bucket="bucket",
    Key="uploads/${filename}",
    Conditions=[
        ["content-length-range", 0, 10 * 1024 * 1024],  # max 10MB
        {"Content-Type": "image/jpeg"},
    ],
    Fields={"Content-Type": "image/jpeg"},
    ExpiresIn=600,
)
# presigned["url"] -- POST URL
# presigned["fields"] -- form fields to include

# Client -- simple copy
s3.copy_object(
    Bucket="dest-bucket",
    Key="dest-key",
    CopySource={"Bucket": "src-bucket", "Key": "src-key"},
)

# Resource -- handles multipart for large objects automatically
s3_resource = boto3.resource("s3")
copy_source = {"Bucket": "src-bucket", "Key": "src-key"}
s3_resource.Object("dest-bucket", "dest-key").copy(copy_source)

s3 = boto3.resource("s3")

# Bucket operations
bucket = s3.Bucket("my-bucket")
for obj in bucket.objects.filter(Prefix="logs/"):
    print(obj.key, obj.size)

# Object operations
obj = s3.Object("my-bucket", "my-key")
obj.upload_file("local.txt")
obj.download_file("local.txt")
obj.delete()

# Read object body
response = obj.get()
data = response["Body"].read()