fal package

Subpackages

Submodules

fal.app module

class fal.app.App(*, _allow_init=False)

Bases: BaseServable

app_auth: ClassVar[Optional[Literal['public', 'private', 'shared']]] = None
app_files: ClassVar[list[str]] = []
app_files_context_dir: ClassVar[Optional[str]] = None
app_files_ignore: ClassVar[list[str]] = ['\\.pyc$', '__pycache__/', '\\.git/', '\\.DS_Store$']
app_name: ClassVar[Optional[str]] = None
collect_routes()
Return type:

dict[RouteSignature, Callable[..., Any]]

concurrency_buffer: ClassVar[int | None] = None
concurrency_buffer_perc: ClassVar[int | None] = None
classmethod get_endpoints()
Return type:

list[str]

health()
host_kwargs: ClassVar[dict[str, Any]] = {'_scheduler': 'nomad', '_scheduler_options': {'storage_region': 'us-east'}, 'keep_alive': 60, 'resolver': 'uv'}
image: ClassVar[Optional[ContainerImage]] = None
isolate_channel: Channel | None = None
kind: ClassVar[Optional[str]] = None
lifespan(app)
local_file_path: ClassVar[Optional[str]] = None
local_python_modules: ClassVar[list[str]] = []
machine_type: ClassVar[str | list[str]] = 'S'
max_concurrency: ClassVar[int | None] = None
max_multiplexing: ClassVar[int | None] = None
min_concurrency: ClassVar[int | None] = None
num_gpus: ClassVar[int | None] = None
provide_hints()

Provide hints for routing the application.

Return type:

list[str]

request_timeout: ClassVar[int | None] = None
requirements: ClassVar[list[str]] = []
setup()

Setup the application before serving.

startup_timeout: ClassVar[int | None] = None
teardown()

Teardown the application after serving.

class fal.app.AppClient(cls, url, timeout=None)

Bases: object

classmethod connect(cls, app_cls, *, health_request_timeout=30, startup_timeout=60, health_check_interval=0.5)
health()
exception fal.app.AppClientError(message, status_code)

Bases: FalServerlessException

message: str
status_code: int
class fal.app.EndpointClient(url, endpoint, signature, timeout=None)

Bases: object

fal.app.endpoint(path, *, is_websocket=False)

Designate the decorated function as an application endpoint.

Return type:

Callable[[TypeVar(EndpointT, bound= Callable[..., Any])], TypeVar(EndpointT, bound= Callable[..., Any])]

async fal.app.open_isolate_channel(address)
Return type:

Channel | None

fal.app.realtime(path, *, buffering=None, session_timeout=None, input_modal=<object object>, max_batch_size=1)

Designate the decorated function as a realtime application endpoint.

Return type:

Callable[[TypeVar(EndpointT, bound= Callable[..., Any])], TypeVar(EndpointT, bound= Callable[..., Any])]

fal.app.wrap_app(cls, **kwargs)
Return type:

IsolatedFunction

fal.apps module

class fal.apps.Completed(logs)

Bases: _Status

Indicates the request has been completed successfully and the result is ready to be retrieved.

logs: list[dict[str, Any]] | None
class fal.apps.InProgress(logs)

Bases: _Status

Indicates the request is now being actively processed, and provides runtime logs for the inference task.

logs: list[dict[str, Any]] | None
class fal.apps.Queued(position)

Bases: _Status

Indicates the request is still in the queue, and provides the position in the queue for ETA calculation.

position: int
class fal.apps.RequestHandle(app_id, request_id, _client=<factory>, _creds=<factory>)

Bases: object

A handle to an async inference request.

app_id: str
cancel()

Cancel an async inference request.

Return type:

None

fetch_result()

Retrieve the result of an async inference request, raises an exception if the request is not completed yet.

Return type:

dict[str, Any]

get()

Retrieve the result of an async inference request, polling the status of the request until it is completed.

Return type:

dict[str, Any]

iter_events(*, logs=False, _RequestHandle__poll_delay=0.2)

Yield all events regarding the given task till its completed.

Return type:

Iterator[_Status]

request_id: str
status(*, logs=False)

Check the status of an async inference request.

Return type:

_Status

fal.apps.run(app_id, arguments, *, path='')

Run an inference task on a Fal app and return the result.

Return type:

dict[str, Any]

fal.apps.stream(app_id, arguments, *, path='')

Stream an inference task on a Fal app.

Return type:

Iterator[str | bytes]

fal.apps.submit(app_id, arguments, *, path='')

Submit an async inference task to the app. Returns a request handle which can be used to check the status of the request and retrieve the result.

Return type:

RequestHandle

fal.apps.ws(app_id, *, path='')

Connect to a HTTP endpoint but with websocket protocol. This is an internal and experimental API, use it at your own risk.

Return type:

Iterator[_WSConnection]

fal.config module

class fal.config.Config(*, validate_profile=False)

Bases: object

DEFAULT_CONFIG_PATH = '~/.fal/config.toml'
delete_profile(profile)
Return type:

None

edit()
Return type:

Iterator[Config]

get(key)
Return type:

Optional[str]

get_internal(key)
Return type:

Optional[str]

property profile: str | None
profiles()
Return type:

List[str]

save()
Return type:

None

set(key, value)
Return type:

None

set_internal(key, value)
Return type:

None

unset(key)
Return type:

None

unset_internal(key)
Return type:

None

fal.container module

class fal.container.ContainerImage(dockerfile_str, build_args=<factory>, registries=<factory>, builder=None, compression='gzip', force_compression=False, secrets=<factory>)

Bases: object

ContainerImage represents a Docker image that can be built from a Dockerfile.

build_args: Dict[str, str]
builder: Optional[Literal['depot', 'service', 'worker']] = None
compression: str = 'gzip'
dockerfile_str: str
force_compression: bool = False
classmethod from_dockerfile(path, **kwargs)
Return type:

ContainerImage

classmethod from_dockerfile_str(text, **kwargs)
Return type:

ContainerImage

registries: Dict[str, Dict[str, str]]
secrets: Dict[str, str]
to_dict()
Return type:

dict

fal.file_sync module

class fal.file_sync.FileMetadata(size, mtime, mode, hash, relative_path, absolute_path)

Bases: object

absolute_path: str
classmethod from_path(file_path, *, relative, absolute)
Return type:

FileMetadata

hash: str
mode: int
mtime: float
relative_path: str
size: int
to_dict()
Return type:

Dict[str, str]

class fal.file_sync.FileSync(local_file_path)

Bases: object

check_hashes_on_server(hashes)
Return type:

List[str]

close()

Close HTTP client.

collect_files(paths, files_context_dir=None)
sync_files(paths, chunk_size=5242880, max_concurrency_uploads=10, files_ignore=[], files_context_dir=None)
Return type:

Tuple[List[FileMetadata], List[AppFileUploadException]]

upload_file_tus(file_path, metadata, chunk_size=5242880)
Return type:

str

fal.file_sync.compute_hash(file_path, mode)
Return type:

str

fal.file_sync.normalize_path(path_str, base_path_str, files_context_dir=None)
Return type:

Tuple[str, str]

fal.file_sync.print_path_tree(file_paths)
fal.file_sync.sanitize_relative_path(rel_path)
Return type:

str

fal.files module

class fal.files.FalFileSystem(*args, **kwargs)

Bases: AbstractFileSystem

get_file(rpath, lpath, **kwargs)

Copy single remote file to local

info(path, **kwargs)

Give details of entry at path

Returns a single dictionary, with exactly the same information as ls would with detail=True.

The default implementation calls ls and could be overridden by a shortcut. kwargs are passed on to `ls().

Some file systems might not be able to measure the file’s size, in which case, the returned dict will include 'size': None.

Returns:

  • dict with keys (name (full path in the FS), size (in bytes), type (file,)

  • directory, or something else) and other FS-specific keys.

ls(path, detail=True, **kwargs)

List objects at path.

This should include subdirectories and files at that location. The difference between a file and a directory must be clear when details are requested.

The specific keys, or perhaps a FileInfo class, or similar, is TBD, but must be consistent across implementations. Must include:

  • full path to the entry (without protocol)

  • size of the entry, in bytes. If the value cannot be determined, will be None.

  • type of entry, “file”, “directory” or other

Additional information may be present, appropriate to the file-system, e.g., generation, checksum, etc.

May use refresh=True|False to allow use of self._ls_from_cache to check for a saved listing and avoid calling the backend. This would be common where listing may be expensive.

Parameters:

  • path (str)

  • detail (bool) – if True, gives a list of dictionaries, where each is the same as the result of info(path). If False, gives a list of paths (str).

  • kwargs (may have additional backend-specific options, such as version) – information

Returns:

  • List of strings if detail is False, or list of directory information

  • dicts if detail is True.

mv(path1, path2, recursive=False, maxdepth=None, **kwargs)

Move file(s) from one location to another

put_file(lpath, rpath, mode='overwrite', **kwargs)

Copy single file to remote

put_file_from_url(url, rpath, mode='overwrite', **kwargs)
rename(path, destination, **kwargs)

Alias of AbstractFileSystem.mv.

rm(path, **kwargs)

Delete files.

Parameters:

  • path (str or list of str) – File(s) to delete.

  • recursive (bool) – If file(s) are directories, recursively delete contents and then also remove the directory

  • maxdepth (int or None) – Depth to pass to walk for finding files to delete, if recursive. If None, there will be no limit and infinite recursion may be possible.

fal.flags module

fal.flags.bool_envvar(name)

fal.project module

fal.project.find_project_root(srcs)

Return a directory containing .git, or pyproject.toml.

That directory will be a common parent of all files and directories passed in srcs.

If no directory in the tree contains a marker that would specify it’s the project root, the root of the file system is returned.

Returns a two-tuple with the first element as the project root path and the second element as a string describing the method by which the project root was discovered.

Return type:

Tuple[Path, str]

fal.project.find_pyproject_toml(path_search_start=None)

Find the absolute filepath to a pyproject.toml if it exists

Return type:

Optional[str]

fal.project.parse_pyproject_toml(path_config)

Parse a pyproject toml file, pulling out relevant parts for fal.

If parsing fails, will raise a tomli.TOMLDecodeError.

Return type:

Dict[str, Any]

fal.rest_client module

class fal.rest_client.CredentialsClient(base_url, *, cookies=NOTHING, headers=NOTHING, timeout=5.0, verify_ssl=True, raise_on_unexpected_status=False, follow_redirects=False)

Bases: Client

get_headers()

Get headers to be used in all endpoints

Return type:

dict[str, str]

fal.sdk module

class fal.sdk.AliasInfo(alias, revision, auth_mode, keep_alive, max_concurrency, max_multiplexing, active_runners, min_concurrency, concurrency_buffer, concurrency_buffer_perc, machine_types, request_timeout, startup_timeout, valid_regions)

Bases: object

active_runners: int
alias: str
auth_mode: str
concurrency_buffer: int
concurrency_buffer_perc: int
keep_alive: int
machine_types: list[str]
max_concurrency: int
max_multiplexing: int
min_concurrency: int
request_timeout: int
revision: str
startup_timeout: int
valid_regions: list[str]
class fal.sdk.ApplicationInfo(application_id, keep_alive, max_concurrency, max_multiplexing, active_runners, min_concurrency, concurrency_buffer, concurrency_buffer_perc, machine_types, request_timeout, startup_timeout, valid_regions, created_at)

Bases: object

active_runners: int
application_id: str
concurrency_buffer: int
concurrency_buffer_perc: int
created_at: datetime
keep_alive: int
machine_types: list[str]
max_concurrency: int
max_multiplexing: int
min_concurrency: int
request_timeout: int
startup_timeout: int
valid_regions: list[str]
class fal.sdk.AuthenticatedCredentials(user=<factory>, team=None)

Bases: Credentials

team: str | None = None
to_grpc()
Return type:

ChannelCredentials

to_headers()
Return type:

dict[str, str]

user: UserAccess
class fal.sdk.Credentials

Bases: object

server_credentials: ServerCredentials = <fal.sdk.RemoteCredentials object>
to_grpc()
Return type:

ChannelCredentials

to_headers()
Return type:

dict[str, str]

class fal.sdk.DeploymentStrategy(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

RECREATE = 'recreate'
ROLLING = 'rolling'
static from_proto(proto)
Return type:

DeploymentStrategy

to_proto()
Return type:

int

class fal.sdk.FalServerlessClient(hostname, credentials=<factory>)

Bases: object

connect()
Return type:

FalServerlessConnection

credentials: Credentials
hostname: str
class fal.sdk.FalServerlessConnection(hostname, credentials, _stack=<factory>, _stub=None)

Bases: object

close()
create_alias(alias, revision, auth_mode)
Return type:

AliasInfo

create_user_key(scope, alias)
Return type:

tuple[str, str]

credentials: Credentials
define_environment(kind, **options)
Return type:

EnvironmentDefinition

delete_alias(alias)
Return type:

str | None

delete_application(application_id)
Return type:

None

delete_secret(name)
Return type:

None

hostname: str
kill_runner(runner_id)
Return type:

None

list_alias_runners(alias, *, list_pending=True, start_time=None)
Return type:

list[RunnerInfo]

list_aliases()
Return type:

list[AliasInfo]

list_applications(application_name=None)
Return type:

list[ApplicationInfo]

list_runners(start_time=None)
Return type:

list[RunnerInfo]

list_secrets()
Return type:

list[ServerlessSecret]

list_user_keys()
Return type:

list[UserKeyInfo]

register(function, environments, application_name=None, auth_mode=None, *, source_code=None, serialization_method='cloudpickle', machine_requirements=None, metadata=None, deployment_strategy, scale=True, private_logs=False, files=None)
Return type:

Iterator[RegisterApplicationResult]

revoke_user_key(key_id)
Return type:

None

run(function, environments, *, serialization_method='cloudpickle', machine_requirements=None, setup_function=None, files=None)
Return type:

Iterator[HostedRunResult[TypeVar(ResultT)]]

scale(application_name, max_concurrency=None)
Return type:

None

set_secret(name, value)
Return type:

None

stop_runner(runner_id)
Return type:

None

property stub: IsolateControllerStub
update_application(application_name, keep_alive=None, max_multiplexing=None, max_concurrency=None, min_concurrency=None, concurrency_buffer=None, concurrency_buffer_perc=None, request_timeout=None, startup_timeout=None, valid_regions=None, machine_types=None)
Return type:

AliasInfo

class fal.sdk.FalServerlessKeyCredentials(key_id, key_secret)

Bases: Credentials

key_id: str
key_secret: str
to_grpc()
Return type:

ChannelCredentials

to_headers()
Return type:

dict[str, str]

class fal.sdk.File(hash, relative_path)

Bases: object

hash: str
relative_path: str
class fal.sdk.HostedRunResult(run_id, status, logs=<factory>, result=None, stream=None, service_urls=None)

Bases: Generic[ResultT]

logs: list[Log]
result: Optional[TypeVar(ResultT)] = None
run_id: str
service_urls: ServiceURLs | None = None
status: HostedRunStatus
stream: Any = None
class fal.sdk.HostedRunState(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

INTERNAL_FAILURE = 2
IN_PROGRESS = 0
SUCCESS = 1
class fal.sdk.HostedRunStatus(state)

Bases: object

state: HostedRunState
class fal.sdk.KeyScope(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

ADMIN = 'ADMIN'
API = 'API'
static from_proto(proto)
Return type:

KeyScope

class fal.sdk.LocalCredentials

Bases: ServerCredentials

to_grpc()
Return type:

ChannelCredentials

class fal.sdk.MachineRequirements(machine_types, num_gpus=None, keep_alive=10, base_image=None, exposed_port=None, scheduler=None, scheduler_options=None, max_concurrency=None, max_multiplexing=None, min_concurrency=None, concurrency_buffer=None, concurrency_buffer_perc=None, request_timeout=None, startup_timeout=None)

Bases: object

base_image: str | None = None
concurrency_buffer: int | None = None
concurrency_buffer_perc: int | None = None
exposed_port: int | None = None
keep_alive: int = 10
machine_types: list[str]
max_concurrency: int | None = None
max_multiplexing: int | None = None
min_concurrency: int | None = None
num_gpus: int | None = None
request_timeout: int | None = None
scheduler: str | None = None
scheduler_options: dict[str, Any] | None = None
startup_timeout: int | None = None
class fal.sdk.RegisterApplicationResult(result, logs=<factory>, service_urls=None)

Bases: object

logs: list[Log]
result: RegisterApplicationResultType | None
service_urls: ServiceURLs | None = None
class fal.sdk.RegisterApplicationResultType(application_id)

Bases: object

application_id: str
class fal.sdk.RemoteCredentials

Bases: ServerCredentials

to_grpc()
Return type:

ChannelCredentials

class fal.sdk.RunnerInfo(runner_id, in_flight_requests, expiration_countdown, uptime, external_metadata, revision, alias, state)

Bases: object

alias: str
expiration_countdown: Optional[int]
external_metadata: dict[str, Any]
in_flight_requests: int
revision: str
runner_id: str
state: RunnerState
uptime: timedelta
class fal.sdk.RunnerState(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

DEAD = 'dead'
DOCKER_PULL = 'docker_pull'
PENDING = 'pending'
RUNNING = 'running'
SETUP = 'setup'
UNKNOWN = 'unknown'
static from_proto(proto)
Return type:

RunnerState

class fal.sdk.ServerCredentials

Bases: object

property base_options: dict[str, str | int]
to_grpc()
Return type:

ChannelCredentials

class fal.sdk.ServerlessSecret(name, created_at)

Bases: object

created_at: datetime
name: str
class fal.sdk.ServiceURLs(playground, run, queue, ws)

Bases: object

playground: str
queue: str
run: str
ws: str
class fal.sdk.UserKeyInfo(key_id, created_at, scope, alias)

Bases: object

alias: str
created_at: datetime
key_id: str
scope: KeyScope
class fal.sdk.WorkerStatus(worker_id, start_time, end_time, duration, user_id, machine_type)

Bases: object

duration: timedelta
end_time: datetime
machine_type: str
start_time: datetime
user_id: str
worker_id: str
fal.sdk.get_agent_credentials(original_credentials)

If running inside a fal Serverless box, use the preconfigured credentials instead of the user provided ones.

Return type:

Credentials

fal.sdk.get_default_credentials(team=None)
Return type:

Credentials

fal.sdk.get_default_server_credentials()
Return type:

ServerCredentials

fal.sync module

fal.sync.sync_dir(local_dir, remote_dir, force_upload=False)
Return type:

str

fal.utils module

class fal.utils.LoadedFunction(function, endpoints, app_name, app_auth, source_code)

Bases: object

app_auth: str | None
app_name: str | None
endpoints: list[str]
function: IsolatedFunction
source_code: str | None
fal.utils.load_function_from(host, file_path, function_name=None)
Return type:

LoadedFunction

fal.workflows module

class fal.workflows.AttributeLeaf(leaf, attribute)

Bases: Leaf

attribute: str
execute(context)
Return type:

Union[Dict[str, Any], List[Any], str, int, float, bool, None, Leaf]

leaf: Leaf
property referee: ReferenceLeaf
class fal.workflows.Context(vars)

Bases: object

hydrate(input)
Return type:

Union[Dict[str, Any], List[Any], str, int, float, bool, None, Leaf]

vars: dict[str, Union[Dict[str, Any], List[Any], str, int, float, bool, None, Leaf]]
class fal.workflows.Display(id, depends, fields)

Bases: Node

execute(context)
Return type:

Union[Dict[str, Any], List[Any], str, int, float, bool, None, Leaf]

fields: list[Leaf]
classmethod from_json(data)
Return type:

Display

to_json()
Return type:

dict[str, Any]

class fal.workflows.IndexLeaf(leaf, index)

Bases: Leaf

execute(context)
Return type:

Union[Dict[str, Any], List[Any], str, int, float, bool, None, Leaf]

index: int
leaf: Leaf
property referee: ReferenceLeaf
class fal.workflows.Leaf

Bases: object

execute(context)
Return type:

Union[Dict[str, Any], List[Any], str, int, float, bool, None, Leaf]

property referee: ReferenceLeaf
exception fal.workflows.MisconfiguredGraphError

Bases: WorkflowSyntaxError

class fal.workflows.Node(id, depends)

Bases: object

depends: set[str]
execute(context)
Return type:

Union[Dict[str, Any], List[Any], str, int, float, bool, None, Leaf]

classmethod from_json(data)
Return type:

Node

id: str
to_json()
Return type:

dict[str, Any]

class fal.workflows.ReferenceLeaf(id)

Bases: Leaf

execute(context)
Return type:

Union[Dict[str, Any], List[Any], str, int, float, bool, None, Leaf]

id: str
property referee: ReferenceLeaf
class fal.workflows.Run(id, depends, app, input)

Bases: Node

app: str
execute(context)
Return type:

Union[Dict[str, Any], List[Any], str, int, float, bool, None, Leaf]

classmethod from_json(data)
Return type:

Run

input: Union[Dict[str, Any], List[Any], str, int, float, bool, None, Leaf]
to_json()
Return type:

dict[str, Any]

class fal.workflows.Workflow(name, input_schema, output_schema, nodes=<factory>, output=None, _app_counter=<factory>)

Bases: object

display(*fields)
Return type:

None

execute(input)
Return type:

Union[Dict[str, Any], List[Any], str, int, float, bool, None, Leaf]

classmethod from_json(data)
Return type:

Workflow

property input: ReferenceLeaf
input_schema: Dict[str, Any]
name: str
nodes: dict[str, Node]
output: dict[str, Any] | None = None
output_schema: Dict[str, Any]
publish(title, *, is_public=True)
run(app, input)
Return type:

ReferenceLeaf

set_output(output)
Return type:

None

to_dict()
Return type:

dict[str, Union[Dict[str, Any], List[Any], str, int, float, bool, None, Leaf]]

to_json()
Return type:

dict[str, Union[Dict[str, Any], List[Any], str, int, float, bool, None, Leaf]]

exception fal.workflows.WorkflowSyntaxError

Bases: FalServerlessException

fal.workflows.create_workflow(name, input, output)
Return type:

Workflow

fal.workflows.depends(data)
Return type:

set[str]

fal.workflows.export_workflow_json(data)
Return type:

Union[Dict[str, Any], List[Any], str, int, float, bool, None, Leaf]

fal.workflows.import_workflow_json(data)
Return type:

Union[Dict[str, Any], List[Any], str, int, float, bool, None, Leaf]

fal.workflows.iter_leaves(data)
Return type:

Iterator[Union[Dict[str, Any], List[Any], str, int, float, bool, None, Leaf]]

fal.workflows.main()
Return type:

None

fal.workflows.parse_leaf(raw_leaf)

Parses a leaf (which is in the form of $variable.field.field_2[index] etc.) into a tree of Leaf objects.

Return type:

Leaf

Module contents

class fal.App(*, _allow_init=False)

Bases: BaseServable

app_auth: ClassVar[Optional[Literal['public', 'private', 'shared']]] = None
app_files: ClassVar[list[str]] = []
app_files_context_dir: ClassVar[Optional[str]] = None
app_files_ignore: ClassVar[list[str]] = ['\\.pyc$', '__pycache__/', '\\.git/', '\\.DS_Store$']
app_name: ClassVar[Optional[str]] = None
collect_routes()
Return type:

dict[RouteSignature, Callable[..., Any]]

concurrency_buffer: ClassVar[int | None] = None
concurrency_buffer_perc: ClassVar[int | None] = None
classmethod get_endpoints()
Return type:

list[str]

health()
host_kwargs: ClassVar[dict[str, Any]] = {'_scheduler': 'nomad', '_scheduler_options': {'storage_region': 'us-east'}, 'keep_alive': 60, 'resolver': 'uv'}
image: ClassVar[Optional[ContainerImage]] = None
isolate_channel: Channel | None = None
kind: ClassVar[Optional[str]] = None
lifespan(app)
local_file_path: ClassVar[Optional[str]] = None
local_python_modules: ClassVar[list[str]] = []
machine_type: ClassVar[str | list[str]] = 'S'
max_concurrency: ClassVar[int | None] = None
max_multiplexing: ClassVar[int | None] = None
min_concurrency: ClassVar[int | None] = None
num_gpus: ClassVar[int | None] = None
provide_hints()

Provide hints for routing the application.

Return type:

list[str]

request_timeout: ClassVar[int | None] = None
requirements: ClassVar[list[str]] = []
setup()

Setup the application before serving.

startup_timeout: ClassVar[int | None] = None
teardown()

Teardown the application after serving.

class fal.ContainerImage(dockerfile_str, build_args=<factory>, registries=<factory>, builder=None, compression='gzip', force_compression=False, secrets=<factory>)

Bases: object

ContainerImage represents a Docker image that can be built from a Dockerfile.

build_args: Dict[str, str]
builder: Optional[Literal['depot', 'service', 'worker']] = None
compression: str = 'gzip'
dockerfile_str: str
force_compression: bool = False
classmethod from_dockerfile(path, **kwargs)
Return type:

ContainerImage

classmethod from_dockerfile_str(text, **kwargs)
Return type:

ContainerImage

registries: Dict[str, Dict[str, str]]
secrets: Dict[str, str]
to_dict()
Return type:

dict

class fal.FalServerlessKeyCredentials(key_id, key_secret)

Bases: Credentials

key_id: str
key_secret: str
to_grpc()
Return type:

ChannelCredentials

to_headers()
Return type:

dict[str, str]

fal.cached(func)

Cache the result of the given function in-memory.

Return type:

Callable[[ParamSpec(ArgsT)], TypeVar(ReturnT, covariant=True)]

fal.endpoint(path, *, is_websocket=False)

Designate the decorated function as an application endpoint.

Return type:

Callable[[TypeVar(EndpointT, bound= Callable[..., Any])], TypeVar(EndpointT, bound= Callable[..., Any])]

fal.function(kind='virtualenv', *, host=None, local_python_modules=None, **config)
fal.realtime(path, *, buffering=None, session_timeout=None, input_modal=<object object>, max_batch_size=1)

Designate the decorated function as a realtime application endpoint.

Return type:

Callable[[TypeVar(EndpointT, bound= Callable[..., Any])], TypeVar(EndpointT, bound= Callable[..., Any])]

fal.sync_dir(local_dir, remote_dir, force_upload=False)
Return type:

str