Amazon S3 Deep Dive — Storage Classes, Versioning, and Python SDK

# s3_operations.py
# Common S3 operations with boto3
import boto3
import json
from io import BytesIO
from datetime import datetime, timedelta

# Initialize client
s3 = boto3.client('s3', region_name='us-east-1')

def list_buckets():
    """List all S3 buckets."""
    response = s3.list_buckets()
    buckets = [b['Name'] for b in response['Buckets']]
    print(f"Buckets: {buckets}")
    return buckets

def upload_file(bucket, key, data):
    """Upload data to S3."""
    s3.put_object(Bucket=bucket, Key=key, Body=data)
    print(f"Uploaded: s3://{bucket}/{key}")

def download_file(bucket, key):
    """Download object from S3."""
    response = s3.get_object(Bucket=bucket, Key=key)
    return response['Body'].read().decode('utf-8')

def list_objects(bucket, prefix=''):
    """List objects in a bucket."""
    response = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)
    objects = response.get('Contents', [])
    print(f"Objects in s3://{bucket}/{prefix}:")
    for obj in objects:
        size_kb = obj['Size'] / 1024
        modified = obj['LastModified'].strftime('%Y-%m-%d %H:%M')
        print(f"  {obj['Key']:<40} {size_kb:>8.1f}KB  {modified}")
    return objects

def generate_presigned_url(bucket, key, expires_in=3600):
    """Generate a presigned URL for temporary access."""
    url = s3.generate_presigned_url(
        'get_object',
        Params={'Bucket': bucket, 'Key': key},
        ExpiresIn=expires_in,
    )
    print(f"Presigned URL (expires in {expires_in}s):")
    print(f"  {url[:80]}...")
    return url

def set_bucket_versioning(bucket, status='Enabled'):
    """Enable or suspend bucket versioning."""
    s3.put_bucket_versioning(
        Bucket=bucket,
        VersioningConfiguration={'Status': status},
    )
    print(f"Versioning {status.lower()} for {bucket}")

def get_bucket_lifecycle(bucket):
    """Get lifecycle rules for a bucket."""
    try:
        response = s3.get_bucket_lifecycle_configuration(Bucket=bucket)
        rules = response.get('Rules', [])
        print(f"Lifecycle rules for {bucket}:")
        for rule in rules:
            print(f"  {rule['ID']}: {rule['Status']}")
    except Exception as e:
        print(f"No lifecycle config: {e}")

# Simulate operations without AWS
class S3Simulator:
    def __init__(self):
        self.buckets = {}

    def create_bucket(self, name):
        self.buckets[name] = {}
        print(f"[S3] Created bucket: {name}")

    def put_object(self, bucket, key, data):
        if bucket not in self.buckets:
            raise ValueError(f"Bucket {bucket} doesn't exist")
        self.buckets[bucket][key] = {
            "data": data,
            "size": len(data),
            "last_modified": datetime.now().isoformat(),
        }
        print(f"[S3] Put: s3://{bucket}/{key} ({len(data)} bytes)")

    def get_object(self, bucket, key):
        obj = self.buckets.get(bucket, {}).get(key)
        return obj

    def generate_url(self, bucket, key, expires=3600):
        return f"https://{bucket}.s3.amazonaws.com/{key}?X-Amz-Expires={expires}"

# Demo
s3_sim = S3Simulator()
print("=== Amazon S3 Operations Demo ===\n")

s3_sim.create_bucket("dodatech-data-lake")
s3_sim.put_object("dodatech-data-lake", "logs/2026/06/15/app.log", "INFO: Server started\nERROR: Connection timeout\nINFO: Retry succeeded")
s3_sim.put_object("dodatech-data-lake", "data/sales_2026.csv", "order_id,amount,status\n1001,250.00,completed\n1002,45.50,pending")

obj = s3_sim.get_object("dodatech-data-lake", "data/sales_2026.csv")
print(f"\nRetrieved object: {obj['size']} bytes")

url = s3_sim.generate_url("dodatech-data-lake", "data/sales_2026.csv")
print(f"Presigned URL: {url}")

print("\n=== S3 Storage Classes Comparison ===")
classes = {
    "S3 Standard": {"cost_per_gb": 0.023, "retrieval": "Instant", "min_days": 0},
    "S3 Standard-IA": {"cost_per_gb": 0.0125, "retrieval": "Instant", "min_days": 30},
    "S3 Glacier Instant": {"cost_per_gb": 0.004, "retrieval": "Instant", "min_days": 90},
    "S3 Glacier Deep Archive": {"cost_per_gb": 0.00099, "retrieval": "12 hours", "min_days": 180},
}
for name, info in classes.items():
    print(f"  {name:<25} ${info['cost_per_gb']:<8} {info['retrieval']:<15} {info['min_days']}d min")

# s3_static_site.py
# Configure S3 for static website hosting

def configure_static_site(bucket_name):
    """Configure an S3 bucket as a static website."""
    s3 = boto3.client('s3')

    # Enable static website hosting
    s3.put_bucket_website(
        Bucket=bucket_name,
        WebsiteConfiguration={
            'IndexDocument': {'Suffix': 'index.html'},
            'ErrorDocument': {'Key': '404.html'},
        }
    )

    # Make bucket public (or use CloudFront + OAI)
    bucket_policy = {
        "Version": "2012-10-17",
        "Statement": [{
            "Sid": "PublicReadGetObject",
            "Effect": "Allow",
            "Principal": "*",
            "Action": "s3:GetObject",
            "Resource": f"arn:aws:s3:::{bucket_name}/*"
        }]
    }
    s3.put_bucket_policy(Bucket=bucket_name, Policy=json.dumps(bucket_policy))

    print(f"Static site ready: http://{bucket_name}.s3-website-us-east-1.amazonaws.com")

# Simulate
print("=== S3 Static Website Hosting ===")
print("""
Files needed:
  index.html   → Homepage
  404.html     → Error page
  style.css    → Styles
  script.js    → Scripts

Steps:
  1. Create bucket with same name as domain (or use subdomain)
  2. Enable "Static website hosting" → set index/error docs
  3. Set bucket policy for public read
  4. Upload files
  5. (Optional) Point Route 53 domain to S3 endpoint
""")

-- S3 Select SQL example
-- Query a CSV in S3 without downloading the whole file
SELECT s.order_id, s.amount
FROM s3object s
WHERE s.status = 'completed'
  AND s.amount > 100
LIMIT 10

# s3_cost_calculator.py
# Estimate monthly S3 costs based on storage class and data volume
from tabulate import tabulate

class S3CostEstimator:
    def __init__(self):
        self.pricing = {
            "S3 Standard": {"per_gb": 0.023, "per_1000_put": 0.005, "per_1000_get": 0.0004},
            "S3 Standard-IA": {"per_gb": 0.0125, "per_1000_put": 0.01, "per_1000_get": 0.001},
            "S3 Glacier Instant": {"per_gb": 0.004, "per_1000_put": 0.02, "per_1000_get": 0.01},
            "S3 Glacier Deep Archive": {"per_gb": 0.00099, "per_1000_put": 0.05, "per_1000_get": 0.10},
        }

    def estimate(self, storage_class, data_gb, puts_per_month=0, gets_per_month=0):
        pricing = self.pricing.get(storage_class)
        if not pricing:
            return None

        storage_cost = data_gb * pricing["per_gb"]
        put_cost = (puts_per_month / 1000) * pricing["per_1000_put"]
        get_cost = (gets_per_month / 1000) * pricing["per_1000_get"]
        total = storage_cost + put_cost + get_cost

        return {
            "class": storage_class,
            "storage_gb": data_gb,
            "storage_cost": round(storage_cost, 2),
            "put_cost": round(put_cost, 2),
            "get_cost": round(get_cost, 2),
            "total": round(total, 2),
        }

calc = S3CostEstimator()
print("=== S3 Monthly Cost Estimate ===")
scenarios = [
    ("S3 Standard", 500, 50000, 200000),
    ("S3 Standard-IA", 2000, 10000, 50000),
    ("S3 Glacier Instant", 10000, 1000, 5000),
]

for cls, gb, puts, gets in scenarios:
    result = calc.estimate(cls, gb, puts, gets)
    print(f"\n{result['class']} ({gb} GB, {puts} puts, {gets} gets):")
    print(f"  Storage: ${result['storage_cost']} | PUT: ${result['put_cost']} | GET: ${result['get_cost']} | Total: ${result['total']}/mo")