or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

async-operations.mdauth-config.mdfile-processing.mdindex.mdrequest-management.mdsync-operations.md
tile.json

tessl/pypi-fal-client

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

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/fal-client@0.7.x

To install, run

npx @tessl/cli install tessl/pypi-fal-client@0.7.0

index.mddocs/

fal-client

A comprehensive Python client library for interacting with machine learning models deployed on the fal.ai platform. It offers both synchronous and asynchronous API interfaces for executing ML models, supporting various input formats including direct file uploads to fal.media CDN and in-memory data URL encoding for latency-sensitive applications.

Package Information

  • Package Name: fal-client
  • Language: Python
  • Installation: pip install fal-client
  • Minimum Python Version: 3.8+

Core Imports

import fal_client

For direct access to client classes:

from fal_client import SyncClient, AsyncClient

For status and handle classes:

from fal_client import Queued, InProgress, Completed, SyncRequestHandle, AsyncRequestHandle

For encoding utilities:

from fal_client import encode, encode_file, encode_image

Basic Usage

import fal_client

# Set your API key as an environment variable
# export FAL_KEY=your-api-key

# Simple synchronous inference
response = fal_client.run("fal-ai/fast-sdxl", arguments={"prompt": "a cute cat, realistic, orange"})
print(response["images"][0]["url"])

# File upload and processing
audio_url = fal_client.upload_file("path/to/audio.wav")
response = fal_client.run("fal-ai/whisper", arguments={"audio_url": audio_url})
print(response["text"])

# Asynchronous execution with queue tracking
import asyncio

async def main():
    handle = await fal_client.submit_async("fal-ai/fast-sdxl", arguments={"prompt": "a cute cat"})
    
    # Monitor progress
    async for event in handle.iter_events(with_logs=True):
        if isinstance(event, fal_client.Queued):
            print(f"Queued. Position: {event.position}")
        elif isinstance(event, fal_client.InProgress):
            print("Processing...")
        elif isinstance(event, fal_client.Completed):
            break
    
    result = await handle.get()
    print(result["images"][0]["url"])

asyncio.run(main())

Architecture

The fal-client library follows a dual-interface architecture:

  • Global Functions: Direct access to fal.ai services through module-level functions (fal_client.run(), fal_client.submit(), etc.)
  • Client Classes: Explicit client instances (SyncClient, AsyncClient) for advanced configuration and lifecycle management
  • Request Handles: Objects that manage the lifecycle of long-running requests with status monitoring and cancellation capabilities
  • Status System: Type-safe status indicators (Queued, InProgress, Completed) for tracking request progress
  • File Management: Integrated CDN upload capabilities and in-memory encoding for various input formats

This design provides both simple function-based usage for quick tasks and sophisticated request management for complex workflows.

Capabilities

Synchronous Operations

Direct execution and queue-based operations with blocking I/O. Includes immediate inference execution, queue submission with status monitoring, CDN file uploads, and request management operations.

def run(application: str, arguments: AnyJSON, *, path: str = "", timeout: float | None = None, hint: str | None = None) -> AnyJSON: ...
def submit(application: str, arguments: AnyJSON, *, path: str = "", hint: str | None = None, webhook_url: str | None = None, priority: Priority | None = None) -> SyncRequestHandle: ...
def subscribe(application: str, arguments: AnyJSON, *, path: str = "", hint: str | None = None, with_logs: bool = False, on_enqueue: callable[[str], None] | None = None, on_queue_update: callable[[Status], None] | None = None, priority: Priority | None = None) -> AnyJSON: ...
def stream(application: str, arguments: AnyJSON, *, path: str = "/stream", timeout: float | None = None) -> Iterator[dict[str, Any]]: ...
def upload(data: bytes | str, content_type: str, file_name: str | None = None) -> str: ...
def upload_file(path: PathLike) -> str: ...
def upload_image(image: "Image.Image", format: str = "jpeg") -> str: ...
def status(application: str, request_id: str, *, with_logs: bool = False) -> Status: ...
def result(application: str, request_id: str) -> AnyJSON: ...
def cancel(application: str, request_id: str) -> None: ...

Synchronous Operations

Asynchronous Operations

Non-blocking async/await operations for concurrent execution. Mirrors synchronous operations but with async coroutines for high-performance applications requiring concurrent model inference and request management.

async def run_async(application: str, arguments: AnyJSON, *, path: str = "", timeout: float | None = None, hint: str | None = None) -> AnyJSON: ...
async def submit_async(application: str, arguments: AnyJSON, *, path: str = "", hint: str | None = None, webhook_url: str | None = None, priority: Priority | None = None) -> AsyncRequestHandle: ...
async def subscribe_async(application: str, arguments: AnyJSON, *, path: str = "", hint: str | None = None, with_logs: bool = False, on_enqueue: callable[[str], None] | None = None, on_queue_update: callable[[Status], None] | None = None, priority: Priority | None = None) -> AnyJSON: ...
async def stream_async(application: str, arguments: AnyJSON, *, path: str = "/stream", timeout: float | None = None) -> AsyncIterator[dict[str, Any]]: ...
async def upload_async(data: bytes | str, content_type: str, file_name: str | None = None) -> str: ...
async def upload_file_async(path: PathLike) -> str: ...
async def upload_image_async(image: "Image.Image", format: str = "jpeg") -> str: ...
async def status_async(application: str, request_id: str, *, with_logs: bool = False) -> Status: ...
async def result_async(application: str, request_id: str) -> AnyJSON: ...
async def cancel_async(application: str, request_id: str) -> None: ...

Asynchronous Operations

Request Management

Status tracking, cancellation, and result retrieval for long-running inference tasks. Provides handle-based request lifecycle management with real-time status updates and event streaming capabilities.

class SyncRequestHandle:
    request_id: str
    def status(self, *, with_logs: bool = False) -> Status: ...
    def iter_events(self, *, with_logs: bool = False, interval: float = 0.1) -> Iterator[Status]: ...
    def cancel(self) -> None: ...
    def get(self) -> AnyJSON: ...

class AsyncRequestHandle:
    request_id: str
    async def status(self, *, with_logs: bool = False) -> Status: ...
    async def iter_events(self, *, with_logs: bool = False, interval: float = 0.1) -> AsyncIterator[Status]: ...
    async def cancel(self) -> None: ...
    async def get(self) -> AnyJSON: ...

Request Management

File Processing

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.

def upload_file(path: PathLike) -> str: ...
def upload_image(image: "Image.Image", format: str = "jpeg") -> str: ...
def encode_file(path: PathLike) -> str: ...
def encode_image(image: "Image.Image", format: str = "jpeg") -> str: ...
def encode(data: str | bytes, content_type: str) -> str: ...

File Processing

Authentication and Configuration

API key management, Google Colab integration, and client configuration. Handles credential discovery from environment variables, Google Colab userdata, and provides custom client initialization with timeout and authentication settings.

class SyncClient:
    def __init__(self, key: str | None = None, default_timeout: float = 120.0): ...

class AsyncClient:
    def __init__(self, key: str | None = None, default_timeout: float = 120.0): ...

def fetch_credentials() -> str: ...
class MissingCredentialsError(Exception): ...

Authentication and Configuration

Types

from typing import Dict, Any, Literal, Iterator, AsyncIterator
from os import PathLike

# Type aliases
AnyJSON = Dict[str, Any]
Priority = Literal["normal", "low"]

class Status: ...

class Queued(Status):
    position: int

class InProgress(Status):
    logs: list[dict[str, Any]] | None

class Completed(Status):
    logs: list[dict[str, Any]] | None
    metrics: dict[str, Any]

class FalClientError(Exception):
    """Raised when fal.ai API returns an error response."""

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