tessl/pypi-fal-client

Python client library for interacting with machine learning models deployed on the fal.ai platform

—

Pending

Overview

Eval results

Files

File Processing

Name: tessl/pypi-fal-client
Author: tessl

File upload to fal.media CDN and in-memory data URL encoding. Supports various file formats with automatic MIME type detection and provides both CDN upload for persistent storage and data URL encoding for immediate use.

Capabilities

CDN File Upload

Upload files to fal.media CDN for persistent storage and use in model inference. The CDN provides fast, globally distributed access to uploaded files.

def upload_file(path: PathLike) -> str:
    """
    Upload a file from the filesystem to fal.media CDN.
    
    Parameters:
    - path: Path to the file to upload (str or PathLike)
    
    Returns:
    str: URL of the uploaded file on fal.media CDN
    """

async def upload_file_async(path: PathLike) -> str:
    """
    Upload a file from the filesystem to fal.media CDN asynchronously.
    
    Parameters:
    - path: Path to the file to upload (str or PathLike)
    
    Returns:
    str: URL of the uploaded file on fal.media CDN
    """

Usage example:

import fal_client

# Upload an audio file
audio_url = fal_client.upload_file("path/to/audio.wav")

# Use in model inference
response = fal_client.run(
    "fal-ai/whisper", 
    arguments={"audio_url": audio_url}
)
print(response["text"])

Image Upload

Upload PIL Image objects directly to fal.media CDN with format specification.

def upload_image(image: "Image.Image", format: str = "jpeg") -> str:
    """
    Upload a PIL Image object to fal.media CDN.
    
    Parameters:
    - image: PIL Image object to upload
    - format: Image format for upload ("jpeg", "png", "webp", etc.)
    
    Returns:
    str: URL of the uploaded image on fal.media CDN
    """

async def upload_image_async(image: "Image.Image", format: str = "jpeg") -> str:
    """
    Upload a PIL Image object to fal.media CDN asynchronously.
    
    Parameters:
    - image: PIL Image object to upload
    - format: Image format for upload ("jpeg", "png", "webp", etc.)
    
    Returns:
    str: URL of the uploaded image on fal.media CDN
    """

Usage example:

from PIL import Image
import fal_client

# Load and process an image
image = Image.open("input.jpg")
image = image.resize((512, 512))

# Upload to CDN
image_url = fal_client.upload_image(image, format="png")

# Use in model inference
response = fal_client.run(
    "fal-ai/image-processor",
    arguments={"image_url": image_url}
)

Binary Data Upload

Upload raw binary data to fal.media CDN with explicit content type specification.

def upload(data: bytes | str, content_type: str) -> str:
    """
    Upload binary data to fal.media CDN.
    
    Parameters:
    - data: The data to upload (bytes or string)
    - content_type: MIME type of the data (e.g., "audio/wav", "image/jpeg")
    
    Returns:
    str: URL of the uploaded file on fal.media CDN
    """

async def upload_async(data: bytes | str, content_type: str) -> str:
    """
    Upload binary data to fal.media CDN asynchronously.
    
    Parameters:
    - data: The data to upload (bytes or string)
    - content_type: MIME type of the data (e.g., "audio/wav", "image/jpeg")
    
    Returns:
    str: URL of the uploaded file on fal.media CDN
    """

Usage example:

import fal_client

# Upload binary audio data
with open("audio.wav", "rb") as f:
    audio_data = f.read()

audio_url = fal_client.upload(audio_data, "audio/wav")

# Use in inference
response = fal_client.run(
    "fal-ai/audio-processor",
    arguments={"audio_url": audio_url}
)

Data URL Encoding

Encode files and data as base64 data URLs for immediate use without CDN upload. This is ideal for small files and latency-sensitive applications.

def encode_file(path: PathLike) -> str:
    """
    Encode a file from the local filesystem to a data URL with inferred content type.
    
    Parameters:
    - path: Path to the file to encode (str or PathLike)
    
    Returns:
    str: Base64-encoded data URL (data:mime/type;base64,...)
    """

def encode_image(image: "Image.Image", format: str = "jpeg") -> str:
    """
    Encode a PIL Image object to a data URL with the specified format.
    
    Parameters:
    - image: PIL Image object to encode
    - format: Image format for encoding ("jpeg", "png", "webp", etc.)
    
    Returns:
    str: Base64-encoded data URL (data:image/format;base64,...)
    """

def encode(data: str | bytes, content_type: str) -> str:
    """
    Encode raw data to a base64 data URL with specified content type.
    
    Parameters:
    - data: The data to encode (string or bytes)
    - content_type: MIME type for the data URL
    
    Returns:
    str: Base64-encoded data URL (data:content_type;base64,...)
    """

Usage examples:

import fal_client
from PIL import Image

# Encode a file directly
audio_data_url = fal_client.encode_file("path/to/audio.wav")

# Use immediately in inference (no CDN upload)
response = fal_client.run(
    "fal-ai/whisper", 
    arguments={"audio_url": audio_data_url}
)

# Encode an image
image = Image.open("image.jpg")
image_data_url = fal_client.encode_image(image, format="png")

# Use in inference
response = fal_client.run(
    "fal-ai/image-processor",
    arguments={"image_url": image_data_url}
)

# Encode raw data
text_data = "Hello, world!"
text_data_url = fal_client.encode(text_data, "text/plain")

File Processing Patterns

Large File Upload vs. Data URL Encoding

Choose the appropriate method based on file size and use case:

import os
import fal_client

def process_file(file_path, model_id, arguments):
    """Smart file processing based on file size."""
    
    file_size = os.path.getsize(file_path)
    
    # Use data URL for small files (< 1MB) for lower latency
    if file_size < 1024 * 1024:  # 1MB
        file_url = fal_client.encode_file(file_path)
        print(f"Using data URL for small file ({file_size} bytes)")
    else:
        # Use CDN upload for larger files
        file_url = fal_client.upload_file(file_path)
        print(f"Uploaded large file ({file_size} bytes) to CDN")
    
    # Use in inference
    arguments["file_url"] = file_url
    return fal_client.run(model_id, arguments=arguments)

Batch File Processing

Process multiple files efficiently with concurrent uploads:

import asyncio
import fal_client

async def process_batch_files(file_paths, model_id):
    """Process multiple files concurrently."""
    
    # Upload all files concurrently
    upload_tasks = [
        fal_client.upload_file_async(path) 
        for path in file_paths
    ]
    file_urls = await asyncio.gather(*upload_tasks)
    
    # Process all uploaded files concurrently
    inference_tasks = [
        fal_client.run_async(model_id, arguments={"file_url": url})
        for url in file_urls
    ]
    results = await asyncio.gather(*inference_tasks)
    
    return list(zip(file_paths, results))

# Usage
file_paths = ["audio1.wav", "audio2.wav", "audio3.wav"]
results = asyncio.run(process_batch_files(file_paths, "fal-ai/whisper"))

for path, result in results:
    print(f"{path}: {result['text']}")

Image Preprocessing and Upload

Preprocess images before uploading for optimal model performance:

from PIL import Image
import fal_client

def preprocess_and_upload_image(image_path, target_size=(512, 512)):
    """Preprocess image and upload to CDN."""
    
    # Load and preprocess image
    image = Image.open(image_path)
    
    # Convert to RGB if needed
    if image.mode != "RGB":
        image = image.convert("RGB")
    
    # Resize while maintaining aspect ratio
    image.thumbnail(target_size, Image.Resampling.LANCZOS)
    
    # Create new image with target size and paste centered
    result = Image.new("RGB", target_size, (255, 255, 255))
    offset = ((target_size[0] - image.width) // 2, 
              (target_size[1] - image.height) // 2)
    result.paste(image, offset)
    
    # Upload processed image
    image_url = fal_client.upload_image(result, format="jpeg")
    
    return image_url

# Usage
processed_image_url = preprocess_and_upload_image("input.jpg")
response = fal_client.run(
    "fal-ai/image-model",
    arguments={"image_url": processed_image_url}
)

Error Handling

Handle common file processing errors:

import fal_client
import os
from PIL import Image

def safe_file_upload(file_path):
    """Upload file with comprehensive error handling."""
    
    try:
        # Check if file exists
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"File not found: {file_path}")
        
        # Check file size (CDN has limits)
        file_size = os.path.getsize(file_path)
        max_size = 100 * 1024 * 1024  # 100MB
        if file_size > max_size:
            raise ValueError(f"File too large: {file_size} bytes (max: {max_size})")
        
        # Upload file
        return fal_client.upload_file(file_path)
        
    except FileNotFoundError as e:
        print(f"File error: {e}")
        return None
    except ValueError as e:
        print(f"Size error: {e}")
        return None
    except Exception as e:
        print(f"Upload error: {e}")
        return None

def safe_image_upload(image_path):
    """Upload image with format validation."""
    
    try:
        # Validate image format
        image = Image.open(image_path)
        if image.format not in ["JPEG", "PNG", "WEBP", "GIF"]:
            raise ValueError(f"Unsupported image format: {image.format}")
        
        return fal_client.upload_image(image)
        
    except Exception as e:
        print(f"Image upload error: {e}")
        return None

Install with Tessl CLI