Authentication and Configuration

def fetch_credentials() -> str:
    """
    Fetch API credentials from available sources.
    
    Checks in order:
    1. FAL_KEY environment variable
    2. FAL_KEY_ID and FAL_KEY_SECRET environment variables (combined as key_id:secret)
    3. Google Colab userdata (if running in Colab)
    
    Returns:
    str: The API key or key_id:secret combination
    
    Raises:
    MissingCredentialsError: If no credentials found in any source
    """

class MissingCredentialsError(Exception):
    """Raised when API credentials cannot be found or are invalid."""

# Simple API key
export FAL_KEY="your-api-key-here"

# Or key ID and secret (for service accounts)
export FAL_KEY_ID="your-key-id"
export FAL_KEY_SECRET="your-key-secret"

import os
import fal_client

# Set credentials programmatically (not recommended for production)
os.environ["FAL_KEY"] = "your-api-key"

# Credentials are automatically discovered
response = fal_client.run("fal-ai/fast-sdxl", arguments={"prompt": "test"})

def is_google_colab() -> bool:
    """
    Check if the current environment is Google Colab.
    
    Returns:
    bool: True if running in Google Colab, False otherwise
    """

def get_colab_token() -> str | None:
    """
    Get API token from Google Colab userdata if available.
    
    Returns:
    str | None: The API token from Colab userdata, or None if not available
    """

# In Google Colab, store your API key in userdata as "FAL_KEY"
# Then use fal_client normally - credentials are auto-discovered
import fal_client

response = fal_client.run("fal-ai/fast-sdxl", arguments={"prompt": "a colab test"})

class SyncClient:
    """Synchronous client for fal.ai API with custom configuration."""
    
    def __init__(self, key: str | None = None, default_timeout: float = 120.0):
        """
        Initialize synchronous client.
        
        Parameters:
        - key: API key (if None, uses fetch_credentials())
        - default_timeout: Default request timeout in seconds
        """

class AsyncClient:
    """Asynchronous client for fal.ai API with custom configuration."""
    
    def __init__(self, key: str | None = None, default_timeout: float = 120.0):
        """
        Initialize asynchronous client.
        
        Parameters:
        - key: API key (if None, uses fetch_credentials())
        - default_timeout: Default request timeout in seconds
        """

import fal_client

# Create client with custom timeout
sync_client = fal_client.SyncClient(default_timeout=60.0)

# Use client methods directly
response = sync_client.run(
    "fal-ai/fast-sdxl", 
    arguments={"prompt": "test"}
)

# Create async client
async_client = fal_client.AsyncClient(
    key="custom-api-key",
    default_timeout=180.0
)

# Use async client
import asyncio

async def main():
    response = await async_client.run(
        "fal-ai/fast-sdxl",
        arguments={"prompt": "async test"}
    )
    return response

result = asyncio.run(main())

# Environment variable for custom service endpoint
# FAL_RUN_HOST: Custom hostname for fal.ai service (default: "fal.run")

# Use custom fal.ai endpoint
export FAL_RUN_HOST="custom.fal.run"
export FAL_KEY="your-api-key"

import os
import fal_client
from pathlib import Path

def load_credentials_from_file():
    """Load credentials from a secure file."""
    
    # Load from .env file
    env_file = Path.home() / ".fal" / "credentials"
    if env_file.exists():
        with open(env_file) as f:
            api_key = f.read().strip()
        os.environ["FAL_KEY"] = api_key
        return api_key
    
    raise fal_client.MissingCredentialsError("No credentials file found")

def setup_client_with_custom_key(api_key):
    """Set up client with explicit API key."""
    
    # Create clients with explicit key (doesn't use environment)
    sync_client = fal_client.SyncClient(key=api_key)
    async_client = fal_client.AsyncClient(key=api_key)
    
    return sync_client, async_client

import os
import fal_client

class FalConfig:
    """Configuration manager for different environments."""
    
    def __init__(self, environment="production"):
        self.environment = environment
        
        # Environment-specific configuration
        if environment == "development":
            os.environ["FAL_RUN_HOST"] = "dev.fal.run"
            self.timeout = 30.0
        elif environment == "staging":
            os.environ["FAL_RUN_HOST"] = "staging.fal.run"
            self.timeout = 60.0
        else:  # production
            os.environ["FAL_RUN_HOST"] = "fal.run"
            self.timeout = 120.0
    
    def get_client(self):
        """Get configured client for the environment."""
        return fal_client.SyncClient(default_timeout=self.timeout)

# Usage
config = FalConfig("development")
client = config.get_client()

import fal_client
import os

def diagnose_auth_issues():
    """Diagnose and report authentication configuration issues."""
    
    print("=== fal-client Authentication Diagnosis ===")
    
    # Check environment variables
    fal_key = os.getenv("FAL_KEY")
    fal_key_id = os.getenv("FAL_KEY_ID")
    fal_key_secret = os.getenv("FAL_KEY_SECRET")
    
    print(f"FAL_KEY set: {'Yes' if fal_key else 'No'}")
    if fal_key:
        print(f"FAL_KEY length: {len(fal_key)}")
    
    print(f"FAL_KEY_ID set: {'Yes' if fal_key_id else 'No'}")
    print(f"FAL_KEY_SECRET set: {'Yes' if fal_key_secret else 'No'}")
    
    # Check Google Colab
    if fal_client.is_google_colab():
        print("Google Colab detected")
        colab_token = fal_client.get_colab_token()
        print(f"Colab token available: {'Yes' if colab_token else 'No'}")
    else:
        print("Not running in Google Colab")
    
    # Test credential fetch
    try:
        credentials = fal_client.fetch_credentials()
        print(f"Credentials successfully fetched (length: {len(credentials)})")
        
        # Test API call
        response = fal_client.run("fal-ai/fast-sdxl", arguments={"prompt": "test"})
        print("API call successful")
        
    except fal_client.MissingCredentialsError as e:
        print(f"Credential error: {e}")
    except Exception as e:
        print(f"API error: {e}")

# Usage for debugging
diagnose_auth_issues()

import fal_client
import os
from contextlib import contextmanager

@contextmanager
def temporary_credentials(api_key):
    """Context manager for temporary API key usage."""
    
    old_key = os.environ.get("FAL_KEY")
    os.environ["FAL_KEY"] = api_key
    
    try:
        yield
    finally:
        if old_key is not None:
            os.environ["FAL_KEY"] = old_key
        else:
            os.environ.pop("FAL_KEY", None)

# Usage
with temporary_credentials("temp-api-key"):
    response = fal_client.run("fal-ai/fast-sdxl", arguments={"prompt": "test"})
    print(response)

# Original credentials restored automatically