Supabase Storage Buckets — File Upload and Management

# create_buckets.py
# Create storage buckets

import os
from supabase import create_client, Client

url = os.getenv("SUPABASE_URL", "https://example.supabase.co")
key = os.getenv("SUPABASE_SERVICE_ROLE_KEY", "your-service-key")
supabase: Client = create_client(url, key)

def create_buckets():
    # Create a public bucket for avatars
    supabase.storage.create_bucket(
        "avatars",
        options={"public": True}
    )
    print("Created public bucket: avatars")
    
    # Create a private bucket for user files
    supabase.storage.create_bucket(
        "user-files",
        options={"public": False}
    )
    print("Created private bucket: user-files")
    
    # Create a bucket for processing output
    supabase.storage.create_bucket(
        "processing-output",
        options={"public": False}
    )
    print("Created private bucket: processing-output")
    
    # List all buckets
    buckets = supabase.storage.list_buckets()
    for bucket in buckets:
        print(f"  Bucket: {bucket.name} (public: {bucket.public})")

create_buckets()

# upload_files.py
# Upload files to Supabase Storage

def upload_file_example():
    supabase = create_client(
        os.getenv("SUPABASE_URL", "https://example.supabase.co"),
        os.getenv("SUPABASE_ANON_KEY", "your-anon-key")
    )
    
    # Upload a file from bytes
    file_bytes = b"ZIP file content would go here..."
    response = supabase.storage.from_("user-files").upload(
        "documents/report.zip",
        file_bytes,
        {"content-type": "application/zip"}
    )
    print(f"Upload response: {response}")
    print(f"Path: documents/report.zip")
    print(f"Size: {len(file_bytes)} bytes")
    
    # Upload with custom metadata
    metadata = {
        "user_id": "user_abc123",
        "original_name": "annual-report.pdf",
        "upload_source": "web-app"
    }
    response = supabase.storage.from_("user-files").upload(
        "reports/annual-report.pdf",
        file_bytes,
        {"content-type": "application/pdf", "metadata": metadata}
    )
    print(f"Uploaded with metadata: {response}")

upload_file_example()

-- Enable RLS on storage.objects (enabled by default in new projects)
-- Allow users to read their own files
CREATE POLICY "Users can read own files"
ON storage.objects FOR SELECT
USING (auth.uid()::text = (metadata->>'user_id'));

-- Allow users to upload their own files
CREATE POLICY "Users can upload own files"
ON storage.objects FOR INSERT
WITH CHECK (auth.uid()::text = (metadata->>'user_id'));

-- Allow users to delete their own files
CREATE POLICY "Users can delete own files"
ON storage.objects FOR DELETE
USING (auth.uid()::text = (metadata->>'user_id'));

# storage_security.py
# Understanding storage RLS

def storage_rls_guide():
    policies = [
        "SELECT: Users can only see files they own",
        "INSERT: Users can only upload files with their user_id in metadata",
        "UPDATE: Users can only modify their own files",
        "DELETE: Users can only delete their own files",
        "Public bucket: Files are readable without authentication",
        "Private bucket: All operations require proper RLS authorization",
    ]
    
    print("Storage RLS Guidelines:")
    for p in policies:
        print(f"  {p}")

storage_rls_guide()

# serve_files.py
# Generate file access URLs

def serve_files_example():
    supabase = create_client(
        os.getenv("SUPABASE_URL", "https://example.supabase.co"),
        os.getenv("SUPABASE_ANON_KEY", "your-anon-key")
    )
    
    # Get public URL (for public buckets)
    public_url = supabase.storage.from_("avatars").get_public_url(
        "user_abc/avatar.png"
    )
    print(f"Public URL: {public_url}")
    
    # Generate signed URL (for private buckets, expires in 1 hour)
    signed_url = supabase.storage.from_("user-files").create_signed_url(
        "documents/report.zip",
        3600  # expiry in seconds
    )
    print(f"Signed URL: {signed_url['signedURL']}")
    print(f"Expires in: 3600 seconds (1 hour)")
    
    # Download a file
    data = supabase.storage.from_("user-files").download(
        "documents/report.zip"
    )
    print(f"Downloaded file: {len(data)} bytes")

serve_files_example()

def document_storage_system():
    features = [
        "Public bucket for profile images",
        "Private bucket for user documents",
        "RLS policies for user-owned files",
        "Signed URLs with 1-hour expiry",
        "File type validation (PDF, images only)",
        "File size limit enforcement",
        "File listing per user",
        "File deletion with owner check",
    ]
    
    print("Document Storage System:")
    for feature in features:
        print(f"  [ ] {feature}")

document_storage_system()