Storage Operations

@property
def storage(self) -> SupabaseStorageClient | AsyncSupabaseStorageClient:
    """
    Storage client providing complete file storage functionality.
    
    Returns:
    Storage client instance (sync or async) with methods for:
    - File upload, download, update, delete operations
    - Bucket creation, listing, and management
    - Access control and permissions
    - Signed URL generation for secure access
    - File metadata and transformations
    """

# Bucket management methods (from storage3)
def list_buckets(self) -> List[dict]:
    """
    List all storage buckets.

    Returns:
    List of bucket information dictionaries

    Raises:
    StorageException: If listing fails
    """

def get_bucket(self, bucket_id: str) -> dict:
    """
    Get information about a specific bucket.

    Parameters:
    - bucket_id: Unique identifier for the bucket

    Returns:
    Bucket information dictionary

    Raises:
    StorageException: If bucket not found
    """

def create_bucket(self, bucket_id: str, options: dict = None) -> dict:
    """
    Create a new storage bucket.

    Parameters:
    - bucket_id: Unique identifier for the new bucket
    - options: Bucket configuration (public, file_size_limit, allowed_mime_types, etc.)

    Returns:
    Created bucket information

    Raises:
    StorageException: If creation fails or bucket exists
    """

def empty_bucket(self, bucket_id: str) -> dict:
    """
    Remove all files from a bucket.

    Parameters:
    - bucket_id: Bucket to empty

    Returns:
    Operation result

    Raises:
    StorageException: If operation fails
    """

def delete_bucket(self, bucket_id: str) -> dict:
    """
    Delete a storage bucket.

    Parameters:
    - bucket_id: Bucket to delete

    Returns:
    Deletion result

    Raises:
    StorageException: If deletion fails or bucket not empty
    """

# List all buckets
buckets = supabase.storage.list_buckets()
for bucket in buckets:
    print(f"Bucket: {bucket['name']}, Public: {bucket['public']}")

# Create a new bucket
bucket = supabase.storage.create_bucket("user-uploads", {
    "public": False,
    "file_size_limit": 1048576,  # 1MB limit
    "allowed_mime_types": ["image/jpeg", "image/png", "application/pdf"]
})

# Get bucket information
bucket_info = supabase.storage.get_bucket("user-uploads")
print(f"Bucket size: {bucket_info['file_size_limit']} bytes")

# Empty and delete bucket
supabase.storage.empty_bucket("old-bucket")
supabase.storage.delete_bucket("old-bucket")

def from_(self, bucket_id: str):
    """
    Select a bucket for file operations.

    Parameters:
    - bucket_id: Name of the storage bucket

    Returns:
    Bucket file client for file operations

    Note: This provides access to file-specific methods
    """

# File upload methods (accessed via supabase.storage.from_("bucket"))
def upload(
    self, 
    path: str, 
    file: Union[str, bytes, IO], 
    file_options: dict = None
) -> dict:
    """
    Upload a file to the bucket.

    Parameters:
    - path: File path within the bucket (including filename)
    - file: File content as string, bytes, or file-like object
    - file_options: Upload options (content_type, cache_control, upsert, etc.)

    Returns:
    Upload result with file metadata

    Raises:
    StorageException: If upload fails
    """

def upload_from_path(
    self,
    path: str,
    local_path: str,
    file_options: dict = None
) -> dict:
    """
    Upload a file from local filesystem.

    Parameters:
    - path: Destination path in bucket
    - local_path: Local file path to upload
    - file_options: Upload options

    Returns:
    Upload result

    Raises:
    StorageException: If upload or file access fails
    """

# Upload bytes data
file_data = b"Hello, World!"
result = supabase.storage.from_("documents").upload(
    "greeting.txt",
    file_data,
    {"content_type": "text/plain"}
)

# Upload from string
content = "This is a text document."
result = supabase.storage.from_("documents").upload(
    "doc.txt",
    content,
    {"content_type": "text/plain"}
)

# Upload from file object
with open("local_file.pdf", "rb") as f:
    result = supabase.storage.from_("documents").upload(
        "uploaded_file.pdf",
        f,
        {"content_type": "application/pdf"}
    )

# Upload from local file path
result = supabase.storage.from_("images").upload_from_path(
    "profile.jpg",
    "/path/to/local/image.jpg",
    {
        "content_type": "image/jpeg",
        "cache_control": "3600",
        "upsert": True  # Overwrite if exists
    }
)

# Upload with metadata
result = supabase.storage.from_("assets").upload(
    "data/report.json",
    '{"report": "data"}',
    {
        "content_type": "application/json",
        "metadata": {
            "author": "system",
            "version": "1.0"
        }
    }
)

def download(self, path: str) -> bytes:
    """
    Download a file from the bucket.

    Parameters:
    - path: File path within the bucket

    Returns:
    File content as bytes

    Raises:
    StorageException: If file not found or download fails
    """

def download_to_path(self, path: str, local_path: str) -> None:
    """
    Download a file directly to local filesystem.

    Parameters:
    - path: File path in bucket
    - local_path: Local destination path

    Raises:
    StorageException: If download fails
    """

# Download file content
content = supabase.storage.from_("documents").download("report.pdf")
with open("local_report.pdf", "wb") as f:
    f.write(content)

# Download directly to file
supabase.storage.from_("images").download_to_path(
    "profile.jpg",
    "/local/path/downloaded_profile.jpg"
)

# Download and process text content
text_content = supabase.storage.from_("documents").download("data.txt")
text = text_content.decode('utf-8')
print(f"File content: {text}")

def list(
    self, 
    path: str = "", 
    options: dict = None
) -> List[dict]:
    """
    List files in the bucket.

    Parameters:
    - path: Directory path to list (default: root)
    - options: Listing options (limit, offset, sort_by, search, etc.)

    Returns:
    List of file/folder information dictionaries

    Raises:
    StorageException: If listing fails
    """

def update(
    self,
    path: str,
    file: Union[str, bytes, IO],
    file_options: dict = None
) -> dict:
    """
    Update/replace an existing file.

    Parameters:
    - path: File path to update
    - file: New file content
    - file_options: Update options

    Returns:
    Update result

    Raises:
    StorageException: If file not found or update fails
    """

def move(self, from_path: str, to_path: str) -> dict:
    """
    Move/rename a file within the bucket.

    Parameters:
    - from_path: Current file path
    - to_path: New file path

    Returns:
    Move operation result

    Raises:
    StorageException: If move fails
    """

def copy(self, from_path: str, to_path: str) -> dict:
    """
    Copy a file within the bucket.

    Parameters:
    - from_path: Source file path
    - to_path: Destination file path

    Returns:
    Copy operation result

    Raises:
    StorageException: If copy fails
    """

def remove(self, paths: List[str]) -> List[dict]:
    """
    Delete one or more files.

    Parameters:
    - paths: List of file paths to delete

    Returns:
    List of deletion results

    Raises:
    StorageException: If deletion fails
    """

# List all files in bucket
files = supabase.storage.from_("documents").list()
for file in files:
    print(f"File: {file['name']}, Size: {file['metadata']['size']} bytes")

# List files in specific directory
files = supabase.storage.from_("images").list("avatars/", {
    "limit": 50,
    "sort_by": {"column": "name", "order": "asc"}
})

# Search for files
files = supabase.storage.from_("documents").list("", {
    "search": "report",
    "limit": 10
})

# Update existing file
with open("updated_document.pdf", "rb") as f:
    result = supabase.storage.from_("documents").update(
        "report.pdf",
        f,
        {"content_type": "application/pdf"}
    )

# Move file to new location
result = supabase.storage.from_("temp").move(
    "draft.txt",
    "published/final.txt"
)

# Copy file
result = supabase.storage.from_("originals").copy(
    "master.jpg",
    "copies/backup.jpg"
)

# Delete single file
result = supabase.storage.from_("temp").remove(["old_file.txt"])

# Delete multiple files
result = supabase.storage.from_("cache").remove([
    "temp1.txt",
    "temp2.txt", 
    "old_data.json"
])

def get_public_url(self, path: str) -> str:
    """
    Get public URL for a file in a public bucket.

    Parameters:
    - path: File path within the bucket

    Returns:
    Public URL string

    Note: Only works for files in public buckets
    """

def create_signed_url(
    self,
    path: str,
    expires_in: int,
    options: dict = None
) -> dict:
    """
    Create a signed URL for temporary access to private files.

    Parameters:
    - path: File path within the bucket
    - expires_in: URL expiration time in seconds
    - options: Additional options (download, transform, etc.)

    Returns:
    Dict with signed URL and expiration info

    Raises:
    StorageException: If URL generation fails
    """

def create_signed_urls(
    self,
    paths: List[str], 
    expires_in: int,
    options: dict = None
) -> List[dict]:
    """
    Create signed URLs for multiple files.

    Parameters:
    - paths: List of file paths
    - expires_in: URL expiration time in seconds
    - options: Additional options

    Returns:
    List of signed URL dictionaries

    Raises:
    StorageException: If URL generation fails
    """

# Get public URL (for public buckets)
public_url = supabase.storage.from_("public-images").get_public_url("logo.png")
print(f"Public URL: {public_url}")

# Create signed URL for private file
signed_url_data = supabase.storage.from_("private-docs").create_signed_url(
    "confidential.pdf",
    3600,  # Expires in 1 hour
    {"download": True}  # Force download
)
print(f"Signed URL: {signed_url_data['signed_url']}")
print(f"Expires at: {signed_url_data['expires_at']}")

# Create signed URLs for multiple files
signed_urls = supabase.storage.from_("user-files").create_signed_urls(
    ["document1.pdf", "document2.pdf", "image.jpg"],
    7200,  # 2 hours
    {"download": False}
)

for url_data in signed_urls:
    print(f"File: {url_data['path']}, URL: {url_data['signed_url']}")

# Generate download URL with transformation (for images)
download_url = supabase.storage.from_("images").create_signed_url(
    "photo.jpg",
    3600,
    {
        "download": True,
        "transform": {
            "width": 300,
            "height": 300,
            "resize": "cover"
        }
    }
)

def get_file_info(self, path: str) -> dict:
    """
    Get detailed metadata for a file.

    Parameters:
    - path: File path within the bucket

    Returns:
    File metadata dictionary with size, type, created_at, etc.

    Raises:
    StorageException: If file not found
    """

# Get file metadata
info = supabase.storage.from_("documents").get_file_info("report.pdf")
print(f"File size: {info['metadata']['size']} bytes")
print(f"Content type: {info['metadata']['mimetype']}")
print(f"Last modified: {info['updated_at']}")
print(f"ETag: {info['metadata']['eTag']}")

# Storage exceptions (from storage3)
class StorageException(Exception):
    """
    Storage operation errors including:
    - File not found
    - Permission denied
    - Bucket full or quota exceeded
    - Invalid file type
    - Network errors
    """

from storage3.utils import StorageException

# Upload error handling
try:
    result = supabase.storage.from_("documents").upload(
        "large_file.pdf",
        large_file_data
    )
except StorageException as e:
    if "file size limit" in str(e).lower():
        print("File is too large")
    elif "mime type" in str(e).lower():
        print("File type not allowed")
    else:
        print(f"Upload failed: {e}")

# Download error handling
try:
    content = supabase.storage.from_("documents").download("missing_file.pdf")
except StorageException as e:
    if "not found" in str(e).lower():
        print("File does not exist")
    else:
        print(f"Download failed: {e}")

# Bucket operation error handling
try:
    supabase.storage.create_bucket("existing-bucket")
except StorageException as e:
    if "already exists" in str(e).lower():
        print("Bucket name already taken")
    else:
        print(f"Bucket creation failed: {e}")

# Efficient file uploads
# Good: Upload with appropriate content type
result = supabase.storage.from_("images").upload(
    "photo.jpg",
    image_data,
    {
        "content_type": "image/jpeg",
        "cache_control": "31536000"  # Cache for 1 year
    }
)

# Batch operations for multiple files
files_to_delete = ["temp1.txt", "temp2.txt", "temp3.txt"]
result = supabase.storage.from_("temp").remove(files_to_delete)

# Use signed URLs for secure access
signed_url = supabase.storage.from_("private").create_signed_url(
    "sensitive.pdf",
    300,  # Short expiration for sensitive files
    {"download": True}
)

# Organize files with proper path structure
# Good: Organized structure
supabase.storage.from_("user-content").upload(
    f"users/{user_id}/documents/{filename}",
    file_data
)

# Handle large files
# For large files, consider chunked uploads or progress tracking
def upload_with_progress(bucket, path, file_data):
    try:
        result = supabase.storage.from_(bucket).upload(path, file_data)
        return result
    except StorageException as e:
        if "timeout" in str(e).lower():
            # Retry or use chunked upload
            pass
        raise