boto3 S3 helper with paginated listing and presigned URLs, tested offline with moto: working versions (tested 2026-10-15)

boto3==1.43.114
moto==5.2.4
pytest==9.1.1

import boto3
from botocore.config import Config

# SigV4 is required for X-Amz-Expires in presigned URLs. The default signer for S3 is SigV2 in some setups.
SIGV4 = Config(signature_version="s3v4")

class S3Helper:
    def __init__(self, bucket: str, region: str = "us-east-1"):
        self.bucket = bucket
        self.s3 = boto3.client("s3", region_name=region, config=SIGV4)

    def upload(self, key: str, data: bytes, content_type: str = "application/octet-stream") -> None:
        self.s3.put_object(Bucket=self.bucket, Key=key, Body=data, ContentType=content_type)

    def list_keys(self, prefix: str = "", page_size: int = 1000) -> list[str]:
        # list_objects_v2 returns at most 1000 keys per page; the paginator follows continuation tokens.
        pages = self.s3.get_paginator("list_objects_v2").paginate(
            Bucket=self.bucket, Prefix=prefix, PaginationConfig={"PageSize": page_size}
        )
        return [obj["Key"] for page in pages for obj in page.get("Contents", [])]

    def presigned_url(self, key: str, expires: int = 300) -> str:
        # Signing is local, so no network call is made.
        return self.s3.generate_presigned_url(
            "get_object", Params={"Bucket": self.bucket, "Key": key}, ExpiresIn=expires
        )

    def delete_prefix(self, prefix: str) -> int:
        keys = self.list_keys(prefix)
        # delete_objects rejects an empty Objects list and accepts at most 1000 keys per call.
        for i in range(0, len(keys), 1000):
            batch = [{"Key": k} for k in keys[i:i + 1000]]
            self.s3.delete_objects(Bucket=self.bucket, Delete={"Objects": batch, "Quiet": True})
        return len(keys)


def create_bucket(bucket: str, region: str = "us-east-1") -> None:
    s3 = boto3.client("s3", region_name=region)
    # us-east-1 rejects a LocationConstraint; every other region requires one.
    if region == "us-east-1":
        s3.create_bucket(Bucket=bucket)
    else:
        s3.create_bucket(Bucket=bucket, CreateBucketConfiguration={"LocationConstraint": region})

from urllib.parse import parse_qs, urlsplit

import pytest
from moto import mock_aws
from s3_helper import S3Helper, create_bucket

BUCKET = "demo-bucket"

@pytest.fixture(autouse=True)
def fake_aws(monkeypatch):
    # Fake credentials so a broken test can never reach a real AWS account.
    for var in ("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_SESSION_TOKEN"):
        monkeypatch.setenv(var, "testing")
    monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
    monkeypatch.delenv("AWS_PROFILE", raising=False)

@mock_aws
def test_upload_and_list_paginates():
    create_bucket(BUCKET)
    h = S3Helper(BUCKET)
    for i in range(5):
        h.upload(f"docs/{i}.txt", b"x", content_type="text/plain")
    # page_size=2 forces three pages, so the paginator must follow tokens.
    assert h.list_keys("docs/", page_size=2) == [f"docs/{i}.txt" for i in range(5)]
    assert h.s3.head_object(Bucket=BUCKET, Key="docs/0.txt")["ContentType"] == "text/plain"

@mock_aws
def test_presigned_url_has_expiry():
    create_bucket(BUCKET)
    url = S3Helper(BUCKET).presigned_url("docs/0.txt", expires=300)
    assert url.startswith(f"https://{BUCKET}.s3")
    # The URL is only valid inside moto's mock, so we check its query parameters rather than fetching it.
    query = parse_qs(urlsplit(url).query)
    assert query["X-Amz-Expires"] == ["300"]
    assert query["X-Amz-Algorithm"] == ["AWS4-HMAC-SHA256"]

@mock_aws
def test_delete_prefix_only_touches_matching_keys():
    create_bucket(BUCKET)
    h = S3Helper(BUCKET)
    for key in ["logs/a.txt", "logs/b.txt", "keep/c.txt"]:
        h.upload(key, b"x")
    assert h.delete_prefix("logs/") == 2
    assert h.list_keys() == ["keep/c.txt"]

@mock_aws
def test_delete_empty_prefix_is_noop():
    # delete_objects with an empty Objects list raises MalformedXML, so the helper must skip the call.
    create_bucket(BUCKET)
    assert S3Helper(BUCKET).delete_prefix("nothing/") == 0

[pytest]
pythonpath = .