Read and write S3 objects¶

S3Adapter creates a boto3 S3 client. Configure AWS credentials through the usual boto3 credential chain or pass explicit values to the constructor.

import requests

from session_adapters.s3_adapter import S3Adapter

session = requests.Session()
session.mount("s3://", S3Adapter(region_name="eu-west-1"))

Download an object¶

response = session.get("s3://example-bucket/reports/latest.json", stream=True)
response.raise_for_status()

with response:
    for chunk in response.iter_content(chunk_size=64 * 1024):
        if chunk:
            process(chunk)

Request a byte range or a particular object version with URL query options:

response = session.get(
    "s3://example-bucket/archive.bin"
    "?range=bytes%3D0-1023&versionId=example-version"
)

Upload an object¶

response = session.put(
    "s3://example-bucket/reports/latest.json",
    data=b'{"state": "ready"}',
    headers={
        "Content-Type": "application/json",
        "Cache-Control": "max-age=60",
    },
)
response.raise_for_status()

To request server-side encryption:

# Amazon S3 managed keys
session.put(
    "s3://example-bucket/private/report.json?sse=AES256",
    data=b"{}",
)

# AWS KMS
session.put(
    "s3://example-bucket/private/report.json"
    "?sse=aws%3Akms&kmsKeyId=example-key-id",
    data=b"{}",
)

List a prefix¶

A GET whose key is empty or ends in / calls ListObjectsV2:

response = session.get(
    "s3://example-bucket/reports/?delimiter=%2F&maxKeys=100"
)
listing = response.json()

for item in listing["Contents"]:
    print(item["Key"], item["Size"])

The response is a compact JSON representation containing KeyCount, IsTruncated, Contents, and CommonPrefixes. It does not expose a continuation-token query option, so one request cannot currently traverse a truncated listing.

Delete an object¶

response = session.delete("s3://example-bucket/reports/obsolete.json")
response.raise_for_status()

See the S3 reference for all mapped headers and query options.