fal.toolkit

from fal.toolkit import (
    Audio,
    CompressedFile,
    File,
    Image,
    KVStore,
    Video,
    FalTookitException,
    FileUploadException,
    KVStoreException,
    DownloadError,
    get_image_size,
    optimize,
    clone_repository,
    download_file,
    download_model_weights,
    get_gpu_type,
    load_inductor_cache,
    sync_inductor_cache,
    synchronized_inductor_cache,
)

Classes

Audio

class fal.toolkit.Audio

Inherits from: File

Constructor Parameters

Name	Type	Default	Description
`__pydantic_self__`	-	-	-
`data`	`Any`	-	-

Methods

Config

def Config()

CompressedFile

class fal.toolkit.CompressedFile

Inherits from: File

Constructor Parameters

Name	Type	Default	Description
`__pydantic_self__`	-	-	-
`data`	`Any`	-	-

Class Variables

Name	Type	Default	Description
`extract_dir`	`Optional[str]`	-	-

Methods

glob

def glob(self, pattern: 'str')

Parameter	Type	Default	Description
`pattern`	`str`	-	-

File

class fal.toolkit.File

Inherits from: BaseModel

Constructor Parameters

Name	Type	Default	Description
`__pydantic_self__`	-	-	-
`data`	`Any`	-	-

Class Variables

Name	Type	Default	Description
`url`	`str`	-	-
`content_type`	`Optional[str]`	-	-
`file_name`	`Optional[str]`	-	-
`file_size`	`Optional[int]`	-	-
`file_data`	`Optional[bytes]`	-	-

Methods

as_bytes

def as_bytes(self) -> 'bytes'

Returns: bytes

from_bytes

def from_bytes(cls, data: 'bytes', content_type: 'Optional[str]' = None, file_name: 'Optional[str]' = None, repository: 'FileRepository | RepositoryId' = 'fal_v3', fallback_repository: 'Optional[FileRepository | RepositoryId | list[FileRepository | RepositoryId]]' = ['cdn', 'fal'], request: 'Optional[Request]' = None, save_kwargs: 'Optional[dict]' = None, fallback_save_kwargs: 'Optional[dict]' = None) -> 'File'

Parameter	Type	Default	Description
`data`	`bytes`	-	-
`content_type`	`Optional[str]`	`None`	-
`file_name`	`Optional[str]`	`None`	-
`repository`	`FileRepository \| Literal[fal, fal_v2, fal_v3, in_memory, gcp_storage, r2, cdn]`	`'fal_v3'`	-
`fallback_repository`	`FileRepository \| Literal[fal, fal_v2, fal_v3, in_memory, gcp_storage, r2, cdn] \| list[FileRepository \| Literal[fal, fal_v2, fal_v3, in_memory, gcp_storage, r2, cdn]] \| NoneType`	`['cdn', 'fal']`	-
`request`	`Optional[Request]`	`None`	-
`save_kwargs`	`Optional[dict]`	`None`	-
`fallback_save_kwargs`	`Optional[dict]`	`None`	-

Returns: File

from_path

def from_path(cls, path: 'str | Path', content_type: 'Optional[str]' = None, repository: 'FileRepository | RepositoryId' = 'fal_v3', multipart: 'bool | None' = None, fallback_repository: 'Optional[FileRepository | RepositoryId | list[FileRepository | RepositoryId]]' = ['cdn', 'fal'], request: 'Optional[Request]' = None, save_kwargs: 'Optional[dict]' = None, fallback_save_kwargs: 'Optional[dict]' = None) -> 'File'

Parameter	Type	Default	Description
`path`	`str \| pathlib.Path`	-	-
`content_type`	`Optional[str]`	`None`	-
`repository`	`FileRepository \| Literal[fal, fal_v2, fal_v3, in_memory, gcp_storage, r2, cdn]`	`'fal_v3'`	-
`multipart`	`bool \| None`	`None`	-
`fallback_repository`	`FileRepository \| Literal[fal, fal_v2, fal_v3, in_memory, gcp_storage, r2, cdn] \| list[FileRepository \| Literal[fal, fal_v2, fal_v3, in_memory, gcp_storage, r2, cdn]] \| NoneType`	`['cdn', 'fal']`	-
`request`	`Optional[Request]`	`None`	-
`save_kwargs`	`Optional[dict]`	`None`	-
`fallback_save_kwargs`	`Optional[dict]`	`None`	-

Returns: File

save

def save(self, path: 'str | Path', overwrite: 'bool' = False) -> 'Path'

Parameter	Type	Default	Description
`path`	`str \| pathlib.Path`	-	-
`overwrite`	`bool`	`False`	-

Returns: Path

Image

class fal.toolkit.Image

Represents an image file.

Inherits from: File

Constructor Parameters

Name	Type	Default	Description
`__pydantic_self__`	-	-	-
`data`	`Any`	-	-

Class Variables

Name	Type	Default	Description
`width`	`Optional[int]`	-	-
`height`	`Optional[int]`	-	-

Methods

Config

def Config()

from_bytes

def from_bytes(cls, data: 'bytes', format: 'ImageFormat', size: 'ImageSize | None' = None, file_name: 'str | None' = None, repository: 'FileRepository | RepositoryId' = 'fal_v3', fallback_repository: 'Optional[FileRepository | RepositoryId | list[FileRepository | RepositoryId]]' = ['cdn', 'fal'], request: 'Optional[Request]' = None) -> 'Image'

Parameter	Type	Default	Description
`data`	`bytes`	-	-
`format`	`Literal[png, jpeg, jpg, webp, gif]`	-	-
`size`	`fal.toolkit.image.image.ImageSize \| None`	`None`	-
`file_name`	`str \| None`	`None`	-
`repository`	`FileRepository \| Literal[fal, fal_v2, fal_v3, in_memory, gcp_storage, r2, cdn]`	`'fal_v3'`	-
`fallback_repository`	`FileRepository \| Literal[fal, fal_v2, fal_v3, in_memory, gcp_storage, r2, cdn] \| list[FileRepository \| Literal[fal, fal_v2, fal_v3, in_memory, gcp_storage, r2, cdn]] \| NoneType`	`['cdn', 'fal']`	-
`request`	`Optional[Request]`	`None`	-

Returns: Image

from_pil

def from_pil(cls, pil_image: 'PILImage.Image', format: 'ImageFormat | None' = None, file_name: 'str | None' = None, repository: 'FileRepository | RepositoryId' = 'fal_v3', fallback_repository: 'Optional[FileRepository | RepositoryId | list[FileRepository | RepositoryId]]' = ['cdn', 'fal'], request: 'Optional[Request]' = None) -> 'Image'

Parameter	Type	Default	Description
`pil_image`	`PILImage.Image`	-	-
`format`	`ImageFormat \| None`	`None`	-
`file_name`	`str \| None`	`None`	-
`repository`	`FileRepository \| RepositoryId`	`'fal_v3'`	-
`fallback_repository`	`Optional[FileRepository \| RepositoryId \| list[FileRepository \| RepositoryId]]`	`['cdn', 'fal']`	-
`request`	`Optional[Request]`	`None`	-

to_pil

def to_pil(self, mode: 'str' = 'RGB') -> 'PILImage.Image'

Parameter	Type	Default	Description
`mode`	`str`	`'RGB'`	-

KVStore

class fal.toolkit.KVStore

A key-value store client for interacting with the FAL KV service.

Constructor Parameters

Name	Type	Default	Description
`db_name`	`str`	-	The name of the database/namespace to use for this KV store.

Properties

Name	Type	Description
`auth_headers`	`dict[str, str]`	-

Methods

get

def get(self, key: str) -> Optional[str]

Retrieve a value from the key-value store.

Parameter	Type	Default	Description
`key`	`str`	-	The key to retrieve the value for.

Returns: Optional[str]

set

def set(self, key: str, value: str) -> None

Store a value in the key-value store.

Parameter	Type	Default	Description
`key`	`str`	-	The key to store the value under.
`value`	`str`	-	The value to store.

Returns: NoneType

Video

class fal.toolkit.Video

Inherits from: File

Constructor Parameters

Name	Type	Default	Description
`__pydantic_self__`	-	-	-
`data`	`Any`	-	-

Methods

Config

def Config()

FalTookitException

class fal.toolkit.FalTookitException

Base exception for all toolkit exceptions

Inherits from: Exception

Constructor Parameters

Name	Type	Default	Description
`args`	-	-	-
`kwargs`	-	-	-

FileUploadException

class fal.toolkit.FileUploadException

Raised when file upload fails

Inherits from: FalTookitException

Constructor Parameters

Name	Type	Default	Description
`args`	-	-	-
`kwargs`	-	-	-

KVStoreException

class fal.toolkit.KVStoreException

Raised when KV store operation fails

Inherits from: FalTookitException

Constructor Parameters

Name	Type	Default	Description
`args`	-	-	-
`kwargs`	-	-	-

DownloadError

class fal.toolkit.DownloadError

Common base class for all non-exit exceptions.

Inherits from: Exception

Constructor Parameters

Name	Type	Default	Description
`args`	-	-	-
`kwargs`	-	-	-

Functions

get_image_size

def get_image_size(source: 'ImageSizeInput') -> 'ImageSize'

Parameter	Type	Default	Description
`source`	`ImageSize \| Literal[square_hd, square, portrait_4_3, portrait_16_9, landscape_4_3, landscape_16_9]`	-	-

Returns: ImageSize

optimize

def optimize(module: 'torch.nn.Module', *, optimization_config: 'dict[str, Any] | None' = None) -> 'torch.nn.Module'

Optimize the given torch module with dynamic compilation and quantization techniques. Only applicable under fal’s cloud environment. Warning: This function is experimental and may not work as expected.

Parameter	Type	Default	Description
`module`	`torch.nn.Module`	-	-
`optimization_config`	`dict[str, Any] \| None`	`None`	-

clone_repository

def clone_repository(https_url: 'str', *, commit_hash: 'str | None' = None, target_dir: 'str | Path | None' = None, repo_name: 'str | None' = None, force: 'bool' = False, include_to_path: 'bool' = False) -> 'Path'

Clones a Git repository from the specified HTTPS URL into a local directory. This function clones a Git repository from the specified HTTPS URL into a local directory. It can also checkout a specific commit if the commit_hash is provided. If a custom target_dir or repo_name is not specified, a predefined directory is used for the target directory, and the repository name is determined from the URL.

Parameter	Type	Default	Description
`https_url`	`str`	-	The HTTPS URL of the Git repository to be cloned.
`commit_hash`	`str \| None`	`None`	The commit hash to checkout after cloning.
`target_dir`	`str \| pathlib.Path \| None`	`None`	The directory where the repository will be saved. If not provided, a predefined directory is used.
`repo_name`	`str \| None`	`None`	The name to be used for the cloned repository directory. If not provided, the repository’s name from the URL is used.
`force`	`bool`	`False`	If `True`, the repository is cloned even if it already exists locally and its commit hash matches the provided commit hash. Defaults to `False`.
`include_to_path`	`bool`	`False`	If `True`, the cloned repository is added to the `sys.path`. Defaults to `False`.

Returns: Path

download_file

def download_file(url: 'str', target_dir: 'str | Path', *, force: 'bool' = False, request_headers: 'dict[str, str] | None' = None, filesize_limit: 'int | None' = None) -> 'Path'

Downloads a file from the specified URL to the target directory. The function downloads the file from the given URL and saves it in the specified target directory, provided it is below the given filesize limit. It also checks whether the local file already exists and whether its content length matches the expected content length from the remote file. If the local file already exists and its content length matches the expected content length from the remote file, the existing file is returned without re-downloading it. If the file needs to be downloaded or if an existing file’s content length does not match the expected length, the function proceeds to download and save the file. It ensures that the target directory exists and handles any errors that may occur during the download process, raising a DownloadError if necessary.

Parameter	Type	Default	Description
`url`	`str`	-	The URL of the file to be downloaded.
`target_dir`	`str \| pathlib.Path`	-	The directory where the downloaded file will be saved. If it’s not an absolute path, it’s treated as a relative directory to “/data”.
`force`	`bool`	`False`	If `True`, the file is downloaded even if it already exists locally and its content length matches the expected content length from the remote file. Defaults to `False`.
`request_headers`	`dict[str, str] \| None`	`None`	A dictionary containing additional headers to be included in the HTTP request. Defaults to `None`.
`filesize_limit`	`int \| None`	`None`	An integer specifying the maximum downloadable size, in megabytes. Defaults to `None`.

Returns: Path Raises:

ValueError: If the provided file_name contains a forward slash (’/’).
DownloadError: If an error occurs during the download process.

download_model_weights

def download_model_weights(url: 'str', force: 'bool' = False, request_headers: 'dict[str, str] | None' = None) -> 'Path'

Downloads model weights from the specified URL and saves them to a predefined directory. This function is specifically designed for downloading model weights and stores them in a predefined directory. It calls the download_file function with the provided URL and the target directory set to a pre-defined location for model weights. The downloaded model weights are saved in this directory, and the function returns the full path to the downloaded weights file.

Parameter	Type	Default	Description
`url`	`str`	-	The URL from which the model weights will be downloaded.
`force`	`bool`	`False`	If `True`, the model weights are downloaded even if they already exist locally and their content length matches the expected content length from the remote file. Defaults to `False`.
`request_headers`	`dict[str, str] \| None`	`None`	A dictionary containing additional headers to be included in the HTTP request. Defaults to `None`.

Returns: Path

get_gpu_type

def get_gpu_type() -> 'str'

Detect the GPU type using nvidia-smi. Example:

>>> gpu_type = get_gpu_type()
>>> print(f"Running on: {gpu_type}")
Running on: H100

Returns: str

load_inductor_cache

def load_inductor_cache(cache_key: 'str') -> 'str'

Load PyTorch Inductor compilation cache from global storage. This function:

Sets TORCHINDUCTOR_CACHE_DIR environment variable
Looks for cached compiled kernels in GPU-specific global storage
Unpacks the cache to local temporary directory
Returns a hash of the unpacked directory for change detection

Example:

>>> dir_hash = load_inductor_cache("flux/2")
Found compilation cache at /data/inductor-caches/H100/flux/2.zip, unpacking...
Cache unpacked successfully.

Parameter	Type	Default	Description
`cache_key`	`str`	-	Unique identifier for this cache (e.g., “flux/2”, “mymodel/v1”)

Returns: str

sync_inductor_cache

def sync_inductor_cache(cache_key: 'str', unpacked_dir_hash: 'str') -> 'None'

Sync updated PyTorch Inductor cache back to global storage. This function:

Checks if the local cache has changed (by comparing hashes)
If changed, creates a zip archive of the new cache
Saves it to GPU-specific global storage

Example:

>>> sync_inductor_cache("flux/2", dir_hash)
No changes in the cache dir, skipping sync.
# or
Changes detected in the cache dir, syncing...

Parameter	Type	Default	Description
`cache_key`	`str`	-	Unique identifier for this cache (same as used in load_inductor_cache)
`unpacked_dir_hash`	`str`	-	Hash returned from load_inductor_cache (for change detection)

Returns: NoneType

synchronized_inductor_cache

def synchronized_inductor_cache(cache_key: 'str') -> 'Iterator[None]'

Context manager to automatically load and sync PyTorch Inductor cache. This wraps load_inductor_cache and sync_inductor_cache for convenience. The cache is loaded on entry and synced on exit (even if an exception occurs). Example:

>>> with synchronized_inductor_cache("mymodel/v1"):
...     self.model = torch.compile(self.model)
...     self.warmup()  # Compilation happens here
# Cache is automatically synced after the with block

Parameter	Type	Default	Description
`cache_key`	`str`	-	Unique identifier for this cache (e.g., “flux/2”, “mymodel/v1”)

Returns: Iterator[]

Client Libraries

Serverless SDK

Platform APIs

fal.toolkit

Classes

Audio

Config

CompressedFile

glob

File

as_bytes

from_bytes

from_path

save

Image

Config

from_bytes

from_pil

to_pil

KVStore

get

set

Video

Config

FalTookitException

FileUploadException

KVStoreException

DownloadError

Functions

get_image_size

optimize

clone_repository

download_file

download_model_weights

get_gpu_type

load_inductor_cache

sync_inductor_cache

synchronized_inductor_cache

Client Libraries

Serverless SDK

Platform APIs

​Classes

​Audio

​Config

​CompressedFile

​glob

​File

​as_bytes

​from_bytes

​from_path

​save

​Image

​Config

​from_bytes

​from_pil

​to_pil

​KVStore

​get

​set

​Video

​Config

​FalTookitException

​FileUploadException

​KVStoreException

​DownloadError

​Functions

​get_image_size

​optimize

​clone_repository

​download_file

​download_model_weights

​get_gpu_type

​load_inductor_cache

​sync_inductor_cache

​synchronized_inductor_cache

Classes

Audio

Config

CompressedFile

glob

File

as_bytes

from_bytes

from_path

save

Image

Config

from_bytes

from_pil

to_pil

KVStore

get

set

Video

Config

FalTookitException

FileUploadException

KVStoreException

DownloadError

Functions

get_image_size

optimize

clone_repository

download_file

download_model_weights

get_gpu_type

load_inductor_cache

sync_inductor_cache

synchronized_inductor_cache