tessl/pypi-fal-client

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

—

Pending

Overview

Eval results

Files

Synchronous Operations

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

Direct execution and queue-based operations with blocking I/O. These functions provide immediate access to fal.ai model inference capabilities with traditional synchronous Python patterns.

Capabilities

Direct Inference Execution

Execute ML model inference directly and return results immediately. This is the simplest way to run inference when you don't need queue tracking or status monitoring.

def run(application: str, arguments: AnyJSON, *, path: str = "", timeout: float | None = None, hint: str | None = None) -> AnyJSON:
    """
    Run an application with the given arguments and return the result directly.
    
    Parameters:
    - application: The fal.ai application ID (e.g., "fal-ai/fast-sdxl")
    - arguments: Dictionary of arguments to pass to the model
    - path: Optional subpath when applicable (default: "")
    - timeout: Request timeout in seconds (default: client default_timeout)
    - hint: Optional runner hint for routing (default: None)
    
    Returns:
    dict: The inference result directly from the model
    """

Usage example:

import fal_client

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

Queue-Based Inference

Submit inference requests to a queue and get a handle for tracking progress. This is ideal for long-running models where you need to monitor status and handle potential queuing delays.

def submit(application: str, arguments: AnyJSON, *, path: str = "", hint: str | None = None, webhook_url: str | None = None, priority: Priority | None = None) -> SyncRequestHandle:
    """
    Submit an inference request to the queue and return a handle for tracking.
    
    Parameters:
    - application: The fal.ai application ID (e.g., "fal-ai/fast-sdxl")
    - arguments: Dictionary of arguments to pass to the model
    - path: Optional subpath when applicable (default: "")
    - hint: Optional runner hint for routing (default: None)
    - webhook_url: Optional webhook URL for notifications (default: None)
    - priority: Request priority ("normal" or "low", default: None)
    
    Returns:
    SyncRequestHandle: Handle for tracking the request
    """

Usage example:

import fal_client

handle = fal_client.submit(
    "fal-ai/fast-sdxl",
    arguments={"prompt": "a detailed landscape"}
)

# Monitor progress
for event in handle.iter_events(with_logs=True):
    if isinstance(event, fal_client.Queued):
        print(f"Queued at position: {event.position}")
    elif isinstance(event, fal_client.InProgress):
        print("Processing...")
    elif isinstance(event, fal_client.Completed):
        break

result = handle.get()
print(result["images"][0]["url"])

Streaming Inference

Subscribe to streaming updates for real-time results. This provides immediate access to streaming updates while the model processes the request.

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:
    """
    Subscribe to streaming updates for an inference request.
    
    Parameters:
    - application: The fal.ai application ID
    - arguments: Dictionary of arguments to pass to the model
    - path: Optional subpath when applicable (default: "")
    - hint: Optional runner hint for routing (default: None)
    - with_logs: Include logs in status updates (default: False)
    - on_enqueue: Callback function called when request is enqueued (default: None)
    - on_queue_update: Callback function called on status updates (default: None)
    - priority: Request priority ("normal" or "low", default: None)
    
    Returns:
    dict: The final inference result after streaming updates complete
    """

Real-time Streaming

Stream inference results in real-time for models that support progressive output generation.

def stream(application: str, arguments: AnyJSON, *, path: str = "/stream", timeout: float | None = None) -> Iterator[dict[str, Any]]:
    """
    Stream inference results in real-time.
    
    Parameters:
    - application: The fal.ai application ID
    - arguments: Dictionary of arguments to pass to the model
    - path: Stream endpoint path (default: "/stream")
    - timeout: Request timeout in seconds (default: None)
    
    Returns:
    Iterator[dict]: Iterator of streaming results
    """

Request Status Operations

Check status, retrieve results, and cancel requests using request IDs.

def status(application: str, request_id: str, *, with_logs: bool = False) -> Status:
    """
    Get the current status of a request.
    
    Parameters:
    - application: The fal.ai application ID
    - request_id: The request ID to check
    - with_logs: Include logs in the status response (default: False)
    
    Returns:
    Status: Current request status (Queued, InProgress, or Completed)
    """

def result(application: str, request_id: str) -> AnyJSON:
    """
    Get the result of a completed request.
    
    Parameters:
    - application: The fal.ai application ID  
    - request_id: The request ID to retrieve results for
    
    Returns:
    dict: The inference result
    """

def cancel(application: str, request_id: str) -> None:
    """
    Cancel a pending or in-progress request.
    
    Parameters:
    - application: The fal.ai application ID
    - request_id: The request ID to cancel
    """

File Upload Operations

Upload files to the fal.media CDN for use in model inference.

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
    
    Returns:
    str: URL of the uploaded file on fal.media CDN
    """

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

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 (default: "jpeg")
    
    Returns:
    str: URL of the uploaded image on fal.media CDN
    """

Usage example:

import fal_client

# Upload a file and use in inference
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"])

Install with Tessl CLI