Skip to content

AWS S3

The utkit.store.aws.s3 module provides a simple interface for interacting with Amazon S3 buckets using boto3.

Installation

Install boto3 as an optional dependency:

pip install "utkit[store]"
# or
pip install boto3

Quick start

import boto3
from utkit.store.aws import AWSConfig, S3Client

# Configure AWS credentials
config = AWSConfig(
    access_key_id="your-access-key",
    secret_access_key="your-secret-key",
    region_name="us-east-1",
)

# Create S3 client
s3_client = config.create_client('s3')
s3 = S3Client(client=s3_client)

try:
    # List buckets
    buckets = s3.list_buckets()
    for bucket in buckets:
        print(f"  {bucket['Name']}")

    # Upload a file
    s3.upload_file('local_file.txt', 'my-bucket', 'remote_file.txt')

    # Download a file
    s3.download_file('my-bucket', 'remote_file.txt', 'local_file.txt')

    # Generate presigned URL (for sharing)
    url = s3.generate_presigned_url('my-bucket', 'remote_file.txt', expiration=3600)
    print(url)
except Exception as e:
    print(f"Error: {e}")

AWSConfig

Configuration for AWS credentials and region.

@dataclass
class AWSConfig:
    access_key_id: str
    secret_access_key: str
    region_name: str

    def create_client(self, service_name: str):
        """Create a boto3 client for the specified AWS service."""
Field Type Description
access_key_id str AWS access key ID
secret_access_key str AWS secret access key
region_name str AWS region (e.g., us-east-1, eu-west-1)

Method: - create_client(service_name) — Creates a boto3 client for the specified service (e.g., 's3', 'dynamodb')

S3Client

Main class for S3 operations.

@dataclass
class S3Client:
    client: Any  # boto3 S3 client object

Methods

list_buckets()

Retrieve the list of existing S3 buckets.

def list_buckets(self) -> list[dict]

Returns: - List of bucket dictionaries with 'Name' and 'CreationDate' keys

Raises: - botocore.exceptions.ClientError on AWS API errors

Example: 

try:
    buckets = s3.list_buckets()
    for bucket in buckets:
        print(f"  {bucket['Name']}")
except Exception as e:
    print(f"Error: {e}")

upload_file(file_name, bucket, object_name=None, extra_args=None)

Upload a file to an S3 bucket. Handles large files automatically.

def upload_file(
    self,
    file_name: str,
    bucket: str,
    object_name: str | None = None,
    extra_args: dict | None = None,
) -> bool

Parameters:

Parameter Type Description
file_name str Path to the local file to upload
bucket str S3 bucket name
object_name str \| None S3 object name (defaults to basename of file_name)
extra_args dict \| None Extra arguments like metadata or ACL

Returns: True on success

Raises: - botocore.exceptions.ClientError on AWS API errors - FileNotFoundError if file does not exist

Example: 

try:
    # Simple upload
    s3.upload_file('local_file.txt', 'my-bucket')

    # With custom object name
    s3.upload_file('local_file.txt', 'my-bucket', 'remote_file.txt')

    # With metadata
    s3.upload_file(
        'file.pdf',
        'my-bucket',
        'documents/file.pdf',
        extra_args={'Metadata': {'author': 'john', 'department': 'sales'}}
    )

    # With ACL
    s3.upload_file(
        'file.txt',
        'my-bucket',
        'public_file.txt',
        extra_args={'ACL': 'public-read'}
    )
except Exception as e:
    print(f"Upload failed: {e}")

upload_fileobj(file_obj, bucket, object_name, extra_args=None)

Upload a file-like object to S3.

def upload_fileobj(
    self,
    file_obj: Any,
    bucket: str,
    object_name: str,
    extra_args: dict | None = None,
) -> bool

Note: The file object must be opened in binary mode ('rb')

Returns: True on success

Raises: - botocore.exceptions.ClientError on AWS API errors

Example: 

try:
    with open('file.txt', 'rb') as f:
        s3.upload_fileobj(f, 'my-bucket', 'remote_file.txt')

    # From memory
    import io
    data = io.BytesIO(b"Hello, World!")
    s3.upload_fileobj(data, 'my-bucket', 'memory_file.txt')
except Exception as e:
    print(f"Upload failed: {e}")

download_file(bucket, object_name, file_name, extra_args=None)

Download a file from S3.

def download_file(
    self,
    bucket: str,
    object_name: str,
    file_name: str,
    extra_args: dict | None = None,
) -> bool

Parameters:

Parameter Type Description
bucket str S3 bucket name
object_name str S3 object name to download
file_name str Local path where file will be saved
extra_args dict \| None Extra arguments

Returns: True on success

Raises: - botocore.exceptions.ClientError on AWS API errors (e.g., object not found)

Example: 

try:
    s3.download_file('my-bucket', 'remote_file.txt', 'local_file.txt')
except Exception as e:
    print(f"Download failed: {e}")

download_fileobj(bucket, object_name, file_obj, extra_args=None)

Download a file from S3 to a file-like object.

def download_fileobj(
    self,
    bucket: str,
    object_name: str,
    file_obj: Any,
    extra_args: dict | None = None,
) -> bool

Returns: True on success

Raises: - botocore.exceptions.ClientError on AWS API errors

Example: 

try:
    with open('file.txt', 'wb') as f:
        s3.download_fileobj('my-bucket', 'remote_file.txt', f)

    # To memory
    import io
    data = io.BytesIO()
    s3.download_fileobj('my-bucket', 'remote_file.txt', data)
    print(data.getvalue())
except Exception as e:
    print(f"Download failed: {e}")

generate_presigned_url(bucket, object_name, expiration=3600)

Generate a presigned URL for downloading an S3 object.

def generate_presigned_url(
    self,
    bucket: str,
    object_name: str,
    expiration: int = 3600,
) -> str

Parameters:

Parameter Type Description
bucket str S3 bucket name
object_name str S3 object name
expiration int URL expiration time in seconds (default: 3600)

Returns: Presigned URL string

Raises: - botocore.exceptions.ClientError on AWS API errors

Example: 

try:
    # 1-hour expiration
    url = s3.generate_presigned_url('my-bucket', 'file.txt')
    print(url)

    # 7-day expiration
    url = s3.generate_presigned_url(
        'my-bucket',
        'file.txt',
        expiration=604800  # 7 days
    )

    # Use with requests
    import requests
    response = requests.get(url)
    with open('downloaded.txt', 'wb') as f:
        f.write(response.content)
except Exception as e:
    print(f"Error: {e}")

generate_presigned_post(bucket, object_name, expiration=3600, fields=None, conditions=None)

Generate a presigned URL for uploading a file via HTTP POST.

def generate_presigned_post(
    self,
    bucket: str,
    object_name: str,
    expiration: int = 3600,
    fields: dict | None = None,
    conditions: list | None = None,
) -> dict

Returns: Dictionary with 'url' and 'fields' keys

Raises: - botocore.exceptions.ClientError on AWS API errors

Example: 

try:
    response = s3.generate_presigned_post('my-bucket', 'upload.txt')

    # Use with requests library
    import requests
    with open('upload.txt', 'rb') as f:
        files = {'file': ('upload.txt', f)}
        http_response = requests.post(
            response['url'],
            data=response['fields'],
            files=files
        )
        print(f"Upload status: {http_response.status_code}")

    # Use in HTML form
    print(f"<form action=\"{response['url']}\" method=\"post\" enctype=\"multipart/form-data\">")
    for key, value in response['fields'].items():
        print(f"  <input type=\"hidden\" name=\"{key}\" value=\"{value}\" />")
    print("  <input type=\"file\" name=\"file\" />")
    print("  <input type=\"submit\" value=\"Upload\" />")
    print("</form>")
except Exception as e:
    print(f"Error: {e}")

delete_object(bucket, object_name)

Delete an object from S3.

def delete_object(self, bucket: str, object_name: str) -> bool

Returns: True on success

Raises: - botocore.exceptions.ClientError on AWS API errors

Example: 

try:
    s3.delete_object('my-bucket', 'file.txt')
except Exception as e:
    print(f"Delete failed: {e}")

head_object(bucket, object_name)

Get metadata about an S3 object without downloading it.

def head_object(self, bucket: str, object_name: str) -> dict

Returns: Dictionary with metadata (ContentLength, LastModified, etc.)

Raises: - botocore.exceptions.ClientError on AWS API errors (e.g., object not found)

Example: 

try:
    metadata = s3.head_object('my-bucket', 'file.txt')
    print(f"Size: {metadata['ContentLength']} bytes")
    print(f"Last modified: {metadata['LastModified']}")
except Exception as e:
    print(f"Error: {e}")

Complete Example

import boto3
from utkit.store.aws import AWSConfig, S3Client
from botocore.exceptions import ClientError

# Setup
config = AWSConfig(
    access_key_id="YOUR_KEY",
    secret_access_key="YOUR_SECRET",
    region_name="us-east-1",
)

s3_client = config.create_client('s3')
s3 = S3Client(client=s3_client)

try:
    # List buckets
    print("Available buckets:")
    buckets = s3.list_buckets()
    for bucket in buckets:
        print(f"  - {bucket['Name']}")

    # Upload
    print("\nUploading file...")
    s3.upload_file(
        'data.csv',
        'my-bucket',
        'uploads/data.csv',
        extra_args={'Metadata': {'source': 'database'}}
    )

    # Generate sharing link
    print("\nGenerating download link...")
    url = s3.generate_presigned_url('my-bucket', 'uploads/data.csv', expiration=86400)
    print(f"Share this link (valid 24 hours): {url}")

    # Check file size
    print("\nFile information:")
    metadata = s3.head_object('my-bucket', 'uploads/data.csv')
    print(f"  Size: {metadata['ContentLength']} bytes")
    print(f"  Modified: {metadata['LastModified']}")

    # Download
    print("\nDownloading file...")
    s3.download_file('my-bucket', 'uploads/data.csv', 'downloaded_data.csv')

    # Cleanup
    print("\nDeleting file...")
    s3.delete_object('my-bucket', 'uploads/data.csv')

except ClientError as e:
    print(f"AWS Error: {e}")
except Exception as e:
    print(f"Error: {e}")