S3-compatible object storage with boto3 in Python

import boto3
from botocore.exceptions import ClientError
from botocore.config import Config
from pathlib import Path
import uuid

class ObjectStorage:
    def __init__(
        self,
        bucket: str,
        endpoint_url: str | None = None,  # None = AWS, set for R2/MinIO
        access_key: str = '',
        secret_key: str = '',
        region: str = 'us-east-1',
    ):
        self.bucket = bucket
        self.client = boto3.client(
            's3',
            endpoint_url=endpoint_url,
            aws_access_key_id=access_key,
            aws_secret_access_key=secret_key,
            region_name=region,
            config=Config(
                signature_version='s3v4',
                retries={'max_attempts': 3, 'mode': 'adaptive'},
            ),
        )

    def upload_file(self, file_data: bytes, content_type: str, prefix: str = '') -> str:
        key = f'{prefix}/{uuid.uuid4()}.{content_type.split("/")[1]}'
        self.client.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=file_data,
            ContentType=content_type,
            # CacheControl='public, max-age=31536000',  # 1 year for immutable files
        )
        return key

    def get_presigned_url(self, key: str, expires_in: int = 3600) -> str:
        return self.client.generate_presigned_url(
            'get_object',
            Params={'Bucket': self.bucket, 'Key': key},
            ExpiresIn=expires_in,
        )

    def delete_file(self, key: str) -> None:
        self.client.delete_object(Bucket=self.bucket, Key=key)

    def file_exists(self, key: str) -> bool:
        try:
            self.client.head_object(Bucket=self.bucket, Key=key)
            return True
        except ClientError as e:
            if e.response['Error']['Code'] == '404':
                return False
            raise

# Config for different providers
# AWS S3
storage = ObjectStorage(bucket='my-bucket', region='us-east-1')

# Cloudflare R2 (S3-compatible, no egress fees)
storage = ObjectStorage(
    bucket='my-bucket',
    endpoint_url=f'https://{account_id}.r2.cloudflarestorage.com',
    access_key=R2_ACCESS_KEY,
    secret_key=R2_SECRET_KEY,
)

# MinIO (self-hosted)
storage = ObjectStorage(
    bucket='my-bucket',
    endpoint_url='http://minio:9000',
    access_key='minioadmin',
    secret_key='minioadmin',
)