​

Table of Contents

• latch.init
• latch.resources.tasks
• latch.resources.workflow
• latch.resources.conditional
• latch.resources.map_tasks
• latch.resources.reference_workflow
• latch.types.init
• latch.types.file
• latch.types.directory
• latch.types.metadata
• latch.types.glob
• latch.types.json
• latch.functions.messages
• latch.functions.operators
• latch.functions.secrets
• latch.ldata.init
• latch.registry.init
• latch.registry.project
• latch.account

​

latch.init

​

latch.resources.tasks

​

Functions

​

get_v100_x1_pod()

def get_v100_x1_pod() -> Pod:

pod = get_v100_x1_pod()

​

get_v100_x4_pod()

def get_v100_x4_pod() -> Pod:

pod = get_v100_x4_pod()

​

get_v100_x8_pod()

def get_v100_x8_pod() -> Pod:

pod = get_v100_x8_pod()

​

_get_large_gpu_pod()

def _get_large_gpu_pod() -> Pod:

pod = _get_large_gpu_pod()

​

_get_small_gpu_pod()

def _get_small_gpu_pod() -> Pod:

pod = _get_small_gpu_pod()

​

_get_large_pod()

def _get_large_pod() -> Pod:

pod = _get_large_pod()

​

_get_medium_pod()

def _get_medium_pod() -> Pod:

pod = _get_medium_pod()

​

_get_small_pod()

def _get_small_pod() -> Pod:

pod = _get_small_pod()

​

_get_l40s_pod(instance_type: str, cpu: int, memory_gib: int, gpus: int) -> Pod

• instance_type: str – taint key/value for the pod
• cpu: int – CPUs requested
• memory_gib: int – memory in GiB
• gpus: int – number of GPUs

def _get_l40s_pod(instance_type: str, cpu: int, memory_gib: int, gpus: int) -> Pod:

pod = _get_l40s_pod("g6e-xlarge", cpu=4, memory_gib=32, gpus=1)

​

custom_memory_optimized_task(cpu: int, memory: int)

• This function is deprecated and will be removed in a future release.
• It raises a deprecation warning.

• cpu: int – number of CPU cores to request
• memory: int – memory in GiB to request

def custom_memory_optimized_task(cpu: int, memory: int):

task = custom_memory_optimized_task(16, 128)

​

custom_task(cpu: Union[Callable, int], memory: Union[Callable, int], *, storage_gib: Union[Callable, int] = 500, timeout: Union[datetime.timedelta, int] = 0, **kwargs)

• cpu: Union[Callable, int] – CPU cores to request (integer or callable for dynamic)
• memory: Union[Callable, int] – memory in GiB to request (integer or callable)
• storage_gib: Union[Callable, int] (default 500) – storage in GiB (integer or callable)
• timeout: Union[datetime.timedelta, int] (default 0) – timeout for the task
• **kwargs – additional keyword arguments

def custom_task(
    cpu: Union[Callable, int],
    memory: Union[Callable, int],
    *,
    storage_gib: Union[Callable, int] = 500,
    timeout: Union[datetime.timedelta, int] = 0,
    **kwargs,
):

task = custom_task(8, 32, storage_gib=200)

​

lustre_setup_task()

def lustre_setup_task():

task = lustre_setup_task()

​

nextflow_runtime_task(cpu: int, memory: int, storage_gib: int = 50)

• cpu: int – CPU cores
• memory: int – memory in GiB
• storage_gib: int – storage in GiB (default 50)

def nextflow_runtime_task(cpu: int, memory: int, storage_gib: int = 50):

task = nextflow_runtime_task(4, 16, 64)

​

_get_l40s_pod(instance_type: str, cpu: int, memory_gib: int, gpus: int) -> Pod

• instance_type: str – taint key/value for the pod
• cpu: int – CPUs requested
• memory_gib: int – memory in GiB
• gpus: int – number of GPUs

def _get_l40s_pod(instance_type: str, cpu: int, memory_gib: int, gpus: int) -> Pod:

pod = _get_l40s_pod("g6e-xlarge", cpu=4, memory_gib=32, gpus=1)

​

g6e_xlarge_task, g6e_2xlarge_task, g6e_4xlarge_task, g6e_8xlarge_task, g6e_12xlarge_task, g6e_16xlarge_task, g6e_24xlarge_task, g6e_48xlarge_task

g6e_xlarge_task = functools.partial(
    task,
    task_config=_get_l40s_pod("g6e-xlarge", cpu=4, memory_gib=32, gpus=1)
)

task = g6e_xlarge_task

​

Classes

​

latch.resources.workflow

​

Functions

​

_generate_metadata()

def _generate_metadata(f: Callable) -> LatchMetadata:

• Parameters
  • f: Callable — Function to generate metadata for.
• Returns
  • LatchMetadata — Generated metadata for the function.

​

_inject_metadata()

def _inject_metadata(f: Callable, metadata: LatchMetadata) -> None:

• Parameters
  • f: Callable — Function to mutate the docstring of.
  • metadata: LatchMetadata — Metadata to inject.
• Returns
  • None

​

workflow()

def workflow(
    metadata: Union[LatchMetadata, Callable],
) -> Union[PythonFunctionWorkflow, Callable]:

• Parameters
  • metadata: Union[LatchMetadata, Callable] — Either a LatchMetadata instance or the function to decorate (when used without parentheses).
• Returns
  • Union[PythonFunctionWorkflow, Callable] — A PythonFunctionWorkflow if used directly or a decorator if used with arguments.

​

nextflow_workflow()

def nextflow_workflow(
    metadata: NextflowMetadata,
) -> Callable[[Callable], PythonFunctionWorkflow]:

• Parameters
  • metadata: NextflowMetadata — Metadata for the Nextflow-style workflow.
• Returns
  • Callable[[Callable], PythonFunctionWorkflow] — A decorator compatible with workflow.

​

Examples

from latch.resources.workflow import workflow

@workflow
def add(a: int, b: int) -> int:
    return a + b

from latch.resources.workflow import workflow
from latch.types.metadata import LatchMetadata

md = LatchMetadata("my_op", None)
md.parameters = {}
@workflow(md)
def multiply(x: int, y: int) -> int:
    return x * y

from latch.resources.workflow import nextflow_workflow
from latch.types.metadata import NextflowMetadata

nfm = NextflowMetadata("nf_op", None)
@nextflow_workflow(nfm)
def process(records: list[dict[str, int]]) -> None:
    pass

​

latch.resources.conditional

​

Module: conditional

​

Functions

​

create_conditional_section()

• The conditional sections can be n-ary with as many elif clauses as desired.
• Outputs from conditional nodes can be consumed, and outputs from other tasks can be passed to conditional nodes.
• Boolean expressions in the condition use & (and) and | (or) operators.
• Unary expressions are not allowed. If a task returns a boolean, use built-in truth checks like result.is_true() or result.is_false().

• name (str): The name of the conditional section, to be shown in Latch Console

• ConditionalSection

def create_conditional_section(name: str) -> ConditionalSection:
    ...

section = create_conditional_section("fractions")

​

latch.resources.map_tasks

​

map_tasks

• Several inputs must run through the same code logic
• Multiple data batches need to be processed in parallel
• Hyperparameter optimization

• task_function: The task to be mapped, to be shown in Latch Console

• A conditional section

@task
def a_mappable_task(a: int) -> str:
    inc = a + 2
    stringified = str(inc)
    return stringified

@task
def coalesce(b: typing.List[str]) -> str:
    coalesced = "".join(b)
    return coalesced

@workflow
def my_map_workflow(a: typing.List[int]) -> str:
    mapped_out = map_task(a_mappable_task)(a=a).with_overrides(
        requests=Resources(mem="300Mi"),
        limits=Resources(mem="500Mi"),
        retries=1,
    )
    coalesced = coalesce(b=mapped_out)
    return coalesced

​

latch.resources.reference_workflow

​

Functions

​

workflow_reference()

• Parameters
  • name (str): The name of the launch plan to reference.
  • version (str): The version of the launch plan to reference.
• Returns: The value returned by reference_launch_plan configured with:
  • project=current_workspace()
  • domain="development"
  • name=name
  • version=version

def workflow_reference(
    name: str,
    version: str,
):
    return reference_launch_plan(
        project=current_workspace(),
        domain="development",
        name=name,
        version=version,
    )

​

Examples

from reference_workflow import workflow_reference

lp = workflow_reference(name="my_workflow", version="v1")

​

latch.types.init

​

Module: latch.types (src/latch/types/init.py)

• LatchDir
• LatchOutputDir
• LatchFile
• LatchOutputFile
• file_glob
• DockerMetadata
• Fork
• ForkBranch
• LatchAppearanceType
• LatchAuthor
• LatchMetadata
• LatchParameter
• LatchRule
• Params
• Section
• Spoiler
• Text

​

Functions

​

file_glob()

# Signature defined in latch.types.glob

​

Classes

​

LatchDir

• Description: Class for directory type representations.
• Methods: Not documented in this module; see the corresponding definitions in the submodule.

​

LatchOutputDir

• Description: Class for output directory representations.
• Methods: Not documented in this module; see the corresponding definitions in the submodule.

​

LatchFile

• Description: Class for file path representations.
• Methods: Not documented in this module; see the corresponding definitions in the submodule.

​

LatchOutputFile

• Description: Class for output file representations.
• Methods: Not documented in this module; see the corresponding definitions in the submodule.

​

DockerMetadata

• Description: Class representing Docker metadata.
• Methods: Not documented in this module; see the corresponding definitions in the submodule.

​

Fork

• Description: Class representing a fork of a project.
• Methods: Not documented in this module; see the corresponding definitions in the submodule.

​

ForkBranch

• Description: Class representing a fork branch.
• Methods: Not documented in this module; see the corresponding definitions in the submodule.

​

LatchAppearanceType

• Description: Class representing appearance type information.
• Methods: Not documented in this module; see the corresponding definitions in the submodule.

​

LatchAuthor

• Description: Class representing an author.
• Methods: Not documented in this module; see the corresponding definitions in the submodule.

​

LatchMetadata

• Description: Class representing latch metadata.
• Methods: Not documented in this module; see the corresponding definitions in the submodule.

​

LatchParameter

• Description: Class representing a parameter.
• Methods: Not documented in this module; see the corresponding definitions in the submodule.

​

LatchRule

• Description: Class representing a rule.
• Methods: Not documented in this module; see the corresponding definitions in the submodule.

​

Params

• Description: Class representing a collection of parameters.
• Methods: Not documented in this module; see the corresponding definitions in the submodule.

​

Section

• Description: Class representing a section.
• Methods: Not documented in this module; see the corresponding definitions in the submodule.

​

Spoiler

• Description: Class representing a spoiler.
• Methods: Not documented in this module; see the corresponding definitions in the submodule.

​

Text

• Description: Class representing text content.
• Methods: Not documented in this module; see the corresponding definitions in the submodule.

• This module solely re-exports names from their respective submodules; no additional behavior is defined here.
• For detailed class definitions and methods, refer to the originating modules where these names are implemented.

​

latch.types.file

• LatchOutputFile is a type alias for LatchFile annotated as an output in Flyte. It is defined as: Annotated[LatchFile, FlyteAnnotation({"output": True})]

​

Classes

​

LatchFile

​

Returning a different file to LatchData.

• Initializes with a local path and an optional remote path.
• Normalizes path input depending on its URI scheme (file, latch, or plain string).
• If a remote_path is provided, it is stored; otherwise a remote path may be inferred from the input.
• If a downloader is supplied via kwargs, the FlyteFile base class is initialized with it; otherwise a downloader function is defined that interacts with Flyte’s context to fetch data lazily.

def __init__(
    self,
    path: Union[str, PathLike],
    remote_path: Optional[Union[str, PathLike]] = None,
    **kwargs,
) -> None:
    ...

• size(self) -> int Returns the size of the remote data via LPath(self.remote_path).size().
• _idempotent_set_path(self, hint: Optional[str] = None) -> None Sets the local path using the Flyte context’s file access if not already set.
• _create_imposters(self) -> None Creates local filesystem placeholders (parent directories and empty file) at the local path.
• local_path (property) -> str Local file path for the environment executing the task.
• remote_path (property) -> Optional[str] Remote URL referencing the object (LatchData or S3).
• __repr__(self) -> str String representation, differing based on whether remote_path is set.
• __str__(self) -> str Simple string representation of the LatchFile.

# Basic usage: create a LatchFile with a local path
lf = LatchFile("./my_file.txt")

# With a remote path
lf_remote = LatchFile("./my_file.txt", "latch:///remote_path.txt")

​

LatchFilePathTransformer

def __init__(self) -> None:
    TypeTransformer.__init__(self, name="LatchFilePath", t=LatchFile)

• to_literal(...) -> Literal Convert a LatchFile to a Flyte Literal containing a Blob with the file URI.

def to_literal(
    self,
    ctx: FlyteContext,
    python_val: LatchFile,
    python_type: type[LatchFile],
    expected: LiteralType,
) -> Literal:
    ...

• to_python_value(...) -> LatchFile Convert a Flyte Literal back to a LatchFile. Behavior depends on the expected Python type:
  • If expected_python_type is PathLike, raise TypeError.
  • If expected_python_type is not a subclass of LatchFile, raise TypeError.
  • If the URI is not remote, return a LatchFile-like instance initialized with the URI.
  • If the URI is remote, create a LatchFile with a local path and a downloader that fetches the data.

def to_python_value(
    self,
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Union[type[LatchFile], PathLike],
) -> LatchFile:
    ...

TypeEngine.register(LatchFilePathTransformer())

• This transformer relies on Flyte context utilities such as ctx.file_access to handle remote/local data paths and downloads.

​

latch.types.directory

​

Module: latch.types.directory

• Exports data structures used for describing directory-related GraphQL responses
• Provides LatchDir, a FlyteDirectory-based representation of a directory with local and remote paths
• Exposes LatchOutputDir as a tagged output type
• Provides LatchDirPathTransformer to convert between Python LatchDir objects and Flyte literals

​

Classes

​

LatchDir

• Description: Represents a directory in the context of a task execution. The local path identifies the directory location on local disk; the remote path identifies a remote location (LatchData or S3). It can be used as a task input or as an output returned from a task.
• Methods and attributes:

​

__init__(self, path: Union[str, PathLike], remote_path: Optional[PathLike] = None, **kwargs)

def __init__(
    self,
    path: Union[str, PathLike],
    remote_path: Optional[PathLike] = None,
    **kwargs,
)

​

_idempotent_set_path(self)

def _idempotent_set_path(self):

​

iterdir(self) -> list[Union[LatchFile, "LatchDir"]]

def iterdir(self) -> list[Union[LatchFile, "LatchDir"]]:

​

size_recursive(self)

def size_recursive(self):

​

_create_imposters(self)

def _create_imposters(self):

​

local_path (property)

@property
def local_path(self) -> str:

​

remote_path (property)

@property
def remote_path(self) -> Optional[str]:

​

__repr__(self)

def __repr__(self):

​

__str__(self)

def __str__(self):

• Notes:
  • The class depends on Flyte context management, remote data accessors, and a remote directory concept used by LatchData.
  • It relies on helper utilities such as format_path and normalize_path and integrates with LatchFile for non-directory children.
• Related aliases and transformers:
  • LatchOutputDir is an alias for a LatchDir annotated as an output in Flyte.
  • LatchDirPathTransformer provides conversion between LatchDir instances and Flyte literals.

​

LatchOutputDir

LatchOutputDir = Annotated[LatchDir, FlyteAnnotation({"output": True})]

• Description: Marks a LatchDir as an output so the console can optimize existence checks for remote paths when needed.

​

LatchDirPathTransformer

class LatchDirPathTransformer(FlyteDirToMultipartBlobTransformer):
    def __init__(self):
        TypeTransformer.__init__(self, name="LatchDirPath", t=LatchDir)

• Methods:

to_literal(self, ctx: FlyteContext, python_val: LatchDir, python_type: type[LatchDir], expected: LiteralType)

def to_literal(
    self,
    ctx: FlyteContext,
    python_val: LatchDir,
    python_type: type[LatchDir],
    expected: LiteralType,
):

to_python_value(self, ctx: FlyteContext, lv: Literal, expected_python_type: Union[type[LatchDir], PathLike]) -> FlyteDirectory

def to_python_value(
    self,
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Union[type[LatchDir], PathLike],
) -> FlyteDirectory:

• Notes:
  • This transformer is registered with the TypeEngine to enable automatic handling of LatchDir types in Flyte workflows.

​

Examples

from latch.types.directory import LatchDir

# Basic local directory
d = LatchDir("./foo")

# With a remote path
d_remote = LatchDir("./foo", "latch:///foo")

# Simple usage inside a task (illustrative; actual task decorators/imports not shown here)
# @task
# def my_task(dir: LatchDir):
#     ...
#     return LatchDir("./foo", "latch:///foo")

• Note: The examples above illustrate basic creation and usage based on the constructor signatures and class behavior described in the code.

​

latch.types.metadata

​

Functions

​

_samplesheet_repr(v: Any) -> str

• Description: If the value is None, returns an empty string; if it is a LatchFile or LatchDir, returns its remote_path; if it is an Enum, returns its value; otherwise returns str(v).
• Signature:

def _samplesheet_repr(v: Any) -> str:
    ...

# Basic usage (conceptual)
_samplesheet_repr(None)          # -> ""
_samplesheet_repr(some_file)      # -> file.remote_path
_samplesheet_repr(SomeEnum.VALUE) # -> "VALUE"

​

default_samplesheet_constructor(samples: List[DC], t: DC, delim: str = ",") -> Path

• Description: Writes a CSV named samplesheet.csv using the field names of the dataclass type t as headers. Values are converted using _samplesheet_repr. Returns the path to the created file.
• Signature:

def default_samplesheet_constructor(samples: List[DC], t: DC, delim: str = ",") -> Path:
    ...

# Basic usage
# Given a list of dataclass instances 'samples' of type 'SomeDataclass'
default_samplesheet_constructor(samples, t=SomeDataclass, delim=",")

​

Classes

​

LatchRule

• Attributes:
  • regex: str — A string regular expression which inputs must match
  • message: str — The message to render when an input does not match the regex
• Methods:
  • dict (property) — Returns a dictionary representation of the instance
  • __post_init__ — Validates that regex is a valid regular expression

​

LatchAppearanceEnum

• Members:
  • line — Value: “line”
  • paragraph — Value: “paragraph”

​

MultiselectOption

• Attributes:
  • name: str — Display name of the option
  • value: object — Associated value for the option

​

Multiselect

• Attributes:
  • options: List[MultiselectOption] — Available options
  • allow_custom: bool — Whether custom entries are allowed

​

LatchAuthor

• Attributes:
  • name: Optional[str] — The name of the author
  • email: Optional[str] — The email of the author
  • github: Optional[str] — A link to the GitHub profile of the author

​

FlowBase

• This is a frozen dataclass serving as a common base for UI flow elements like Section, Text, Params, etc.

​

Section

• Attributes:
  • section: str — Title of the section
  • flow: List[FlowBase] — Flow displayed in the section card
• Methods:
  • __init__(section: str, *flow: FlowBase) — Initializes a section with a title and child flow elements

​

Text

• Attributes:
  • text: str — Markdown body text

​

Title

• Attributes:
  • title: str — Markdown title text

​

Params

• Attributes:
  • params: List[str] — Names of parameters whose widgets will be displayed
• Methods:
  • __init__(*args: str) — Initializes with a list of parameter names

​

Spoiler

• Attributes:
  • spoiler: str — Title of the spoiler
  • flow: List[FlowBase] — Flow displayed in the spoiler card
• Methods:
  • __init__(spoiler: str, *flow: FlowBase) — Initializes a spoiler with a title and child flow

​

ForkBranch

• Attributes:
  • display_name: str — String displayed in the fork’s multibutton
  • flow: List[FlowBase] — Child flow displayed when the branch is active
• Methods:
  • __init__(display_name: str, *flow: FlowBase) — Initializes a fork branch

​

Fork

• Attributes:
  • fork: str — Name of a str-typed parameter to store the active branch’s key
  • display_name: str — Title shown above the fork selector
  • flows: Dict[str, ForkBranch] — Mapping between branch keys to branch definitions
• Methods:
  • __init__(fork: str, display_name: str, **flows: ForkBranch) — Initializes fork with branches

​

LatchParameter

• Attributes:
  • display_name: Optional[str] — The name used to display the parameter
  • description: Optional[str] — The parameter’s description
  • hidden: bool — Whether the parameter should be hidden by default
  • section_title: Optional[str] — Section header for grouping
  • placeholder: Optional[str] — Placeholder text for input
  • comment: Optional[str] — Comment about the parameter
  • output: bool — Whether the parameter is an output
  • batch_table_column: bool — Show in batch table in UI
  • allow_dir: bool — Allow directories in UI
  • allow_file: bool — Allow files in UI
  • appearance_type: LatchAppearance — How the parameter should be rendered (line/paragraph or multiselect)
  • rules: List[LatchRule] — Validation rules for inputs
  • detail: Optional[str] — Additional detail
  • samplesheet: Optional[bool] — Use samplesheet input UI
  • samplesheet-related fields:
    • allowed_tables: Optional[List[int]] — Registry Tables allowed for samplesheet
  • _custom_ingestion: Optional[str] — Custom ingestion hook
• Methods:
  • __str__() — YAML-like string representation including metadata
  • dict (property) — Returns a dictionary with an __metadata__ key containing metadata, including nested appearance details and rules

​

SnakemakeParameter(Generic[T], LatchParameter)

• Attributes:
  • type: Optional[Type[T]] — The python type of the parameter
  • default: Optional[T] — Default value
• Inherits all fields from LatchParameter

​

SnakemakeFileParameter(SnakemakeParameter[Union[LatchFile, LatchDir]])

• Attributes:
  • type: Optional[Union[Type[LatchFile], Type[LatchDir]]] — The python type of the parameter
  • path: Optional[Path] — Destination path for file
  • config: bool — Expose path in Snakemake config
  • download: bool — Download in JIT step

​

SnakemakeFileMetadata

• Attributes:
  • path: Path — Local path where the file will be copied
  • config: bool — Expose in Snakemake config
  • download: bool — Download in JIT step

​

NextflowParameter(Generic[T], LatchParameter)

• Attributes:
  • type: Optional[Type[T]] — The python type of the parameter
  • default: Optional[T] — Default value
  • samplesheet_type: Literal["csv", "tsv", None] — Samplesheet type (CSV/TSV)
  • samplesheet_constructor: Optional[Callable[[T], Path]] — Custom samplesheet constructor
  • results_paths: Optional[List[Path]] — Output sub-paths exposed in UI under “Results”
• Methods:
  • __post_init__() — Validates samplesheet constraints and requires a constructor if needed or raises errors via Click

​

DC (TypeVar)

• Used for typing in default_samplesheet_constructor with generic dataclass types

​

_samplesheet_repr(v: Any) -> str (see Functions)

​

default_samplesheet_constructor (see Functions)

​

NextflowRuntimeResources

• Attributes:
  • cpus: Optional[int] — Number of CPUs required for the task
  • memory: Optional[int] — Memory required (GiB)
  • storage_gib: Optional[int] — Storage required (GiB)
  • storage_expiration_hours: int — Hours to retain workdir after failure

​

LatchMetadata

• Attributes:
  • display_name: str — The human-readable name of the workflow
  • author: LatchAuthor — Metadata about the workflow author
  • documentation: Optional[str] — Link to workflow documentation
  • repository: Optional[str] — Link to repository hosting the workflow
  • license: str — SPDX identifier
  • parameters: Dict[str, LatchParameter] — Parameter metadata map
  • wiki_url: Optional[str] — Wiki URL
  • video_tutorial: Optional[str] — Video tutorial URL
  • tags: List[str] — Tags for categorization
  • flow: List[FlowBase] — Flow elements describing UI flow
  • no_standard_bulk_execution: bool — Disable standard bulk execution
  • _non_standard: Dict[str, object] — Non-standard metadata
  • about_page_path: Optional[Path] — Path to markdown about page
• Methods:
  • validate() — Validates fields (e.g., about page path type)
  • dict (property) — Returns a dictionary with metadata; excludes parameters
  • __str__() — YAML-like string with parameters expanded

​

DockerMetadata

• Attributes:
  • username: str — Account username
  • secret_name: str — Secret name containing the password

​

EnvironmentConfig

• Attributes:
  • use_conda: bool — Use Snakemake conda directive
  • use_container: bool — Use Snakemake container directive
  • container_args: List[str] — Additional container arguments

​

FileMetadata (TypeAlias)

• Type: Dict[str, Union[SnakemakeFileMetadata, "FileMetadata"]]

​

SnakemakeMetadata(LatchMetadata)

• Attributes:
  • output_dir: Optional[LatchDir] — Directory for Snakemake outputs
  • name: Optional[str] — Name of the workflow
  • docker_metadata: Optional[DockerMetadata] — Docker credentials
  • env_config: EnvironmentConfig — Environment configuration
  • parameters: Dict[str, SnakemakeParameter] — Snakemake parameter metadata
  • file_metadata: FileMetadata — File metadata mappings
  • cores: int — Number of cores (Snakemake)
  • about_page_content: Optional[Path] — Path to About page content
• Methods:
  • validate() — Validates fields (e.g., about_page_content type)
  • dict (property) — Returns a dictionary including metadata; omits about_page_content
  • __post_init__() — Calls validate(), assigns default workflow name if needed, and registers global metadata

​

NextflowMetadata(LatchMetadata)

• Attributes:
  • name: Optional[str] — Name of the workflow
  • parameters: Dict[str, NextflowParameter] — Nextflow parameter metadata
  • runtime_resources: NextflowRuntimeResources — Runtime resources
  • execution_profiles: List[str] — Execution config profiles
  • log_dir: Optional[LatchDir] — Directory to dump Nextflow logs
  • upload_command_logs: bool — Upload .command.* logs after task execution
• Methods:
  • dict (property) — Returns a dictionary including metadata, omitting about_page_path
  • __post_init__() — Calls validate(), assigns name if missing, and registers global metadata

​

Global Registries

• _snakemake_metadata: Optional[SnakemakeMetadata] — Registry for Snakemake metadata
• _nextflow_metadata: Optional[NextflowMetadata] — Registry for Nextflow metadata

• Many classes include docstrings directly in the code describing their purpose and usage.
• Some fields and behaviors are implemented with validation logic in __post_init__ or validate() methods; these validations are reflected in the class descriptions above where relevant.

​

latch.types.glob

​

Module description

​

Functions

​

file_glob()

• Description Constructs a list of LatchFile objects from a glob pattern. This utility helps pass collections of files between tasks. The remote location of each constructed LatchFile is created by appending the file name matched by the pattern to the directory represented by remote_directory.
• Parameters
  • pattern: str
    A glob pattern to match a set of files, e.g., '*.py'. The pattern is resolved relative to the working directory of the caller unless a target_dir is provided.
  • remote_directory: str
    A valid latch URL pointing to a directory, e.g., 'latch:///foo'. This must be a directory and not a file.
  • target_dir: Optional[Path]
    An optional Path object to define an alternate working directory for path resolution.
• Returns
  • List[LatchFile]
    A list of instantiated LatchFile objects.

def file_glob(
    pattern: str, remote_directory: str, target_dir: Optional[Path] = None
) -> List[LatchFile]:

• Examples

# Basic usage
files = file_glob("*.fastq.gz", "latch:///fastqc_outputs")

​

latch.types.json

​

Module: latch.types.json

• JsonArray
• JsonObject
• JsonValue

​

JsonArray

JsonArray: TypeAlias = List["JsonValue"]
"""JSON-compatible list"""

• Description: JSON-compatible list
• Type alias: List["JsonValue"]

​

JsonObject

JsonObject: TypeAlias = Dict[str, "JsonValue"]
"""JSON-compatible dictionary"""

• Description: JSON-compatible dictionary
• Type alias: Dict[str, "JsonValue"]

​

JsonValue

JsonValue: TypeAlias = Union[JsonObject, JsonArray, str, int, float, bool, None]
"""JSON-compatible value

Can be a dictionary, an array, or a primitive value
"""

• Description: JSON-compatible value. Can be a dictionary, an array, or a primitive value
• Type alias: Union[JsonObject, JsonArray, str, int, float, bool, None]

​

latch.functions.messages

• Exports:
  • NUCLEUS_URL (str): URL derived from environment variable LATCH_CLI_NUCLEUS_URL or a default.
  • ADD_MESSAGE_ENDPOINT (str): Endpoint to post task execution messages.
  • message(typ: str, data: Dict[str, Any]) -> None: Post or print a task execution message.

​

Functions

​

message(typ: str, data: Dict[str, Any])

• Parameters
  • typ: str
    • A message type that determines how your message is displayed. Currently one of 'info', 'warning', or 'error'.
  • data: Dict[str, Any]
    • The data displayed on the Latch console, formatted as follows: {'title': ..., 'body': ...}.
• Returns
  • None
• Raises
  • RuntimeError: If an internal error occurs while processing the message.
• Example usage:

@small_task
def task():

    ...

    try:
        ...
    catch ValueError:
        title = 'Invalid sample ID column selected'
        body = 'Your file indicates that sample columns a, b are valid'
        message(type='error', data={'title': title, 'body': body})

    ...

• Signature (for reference)

def message(typ: str, data: Dict[str, Any]) -> None:

​

latch.functions.operators

​

operators.py

​

Functions

​

_combine(item1: Any, item2: Any)

def _combine(item1: Any, item2: Any):

• If both items are lists, the lists are concatenated
• If one of the items is a list, the other is appended to that list
• Otherwise, the output is a new list containing both items

• item1: Any
• item2: Any

• List[Any] A new list containing the combined items following the rules above.

​

left_join(left: Dict[str, Any], right: Dict[str, Any]) -> Dict[str, Any]

def left_join(left: Dict[str, Any], right: Dict[str, Any]) -> Dict[str, Any]:

• left: Dict[str, Any]
• right: Dict[str, Any]

• Dict[str, Any] Dictionary containing joined values or original left values when no match.

​

right_join(left: Dict[str, Any], right: Dict[str, Any]) -> Dict[str, Any]

def right_join(left: Dict[str, Any], right: Dict[str, Any]) -> Dict[str, Any]:

• left: Dict[str, Any]
• right: Dict[str, Any]

• Dict[str, Any]

​

inner_join(left: Dict[str, Any], right: Dict[str, Any]) -> Dict[str, Any]

def inner_join(left: Dict[str, Any], right: Dict[str, Any]) -> Dict[str, Any]:

• left: Dict[str, Any]
• right: Dict[str, Any]

• Dict[str, Any] Dictionary containing values for keys present in both input dictionaries.

​

outer_join(left: Dict[str, Any], right: Dict[str, Any]) -> Dict[str, Any]

def outer_join(left: Dict[str, Any], right: Dict[str, Any]) -> Dict[str, Any]:

• left: Dict[str, Any]
• right: Dict[str, Any]

• Dict[str, Any] Dictionary containing all keys from both inputs, combining where present in both.

​

group_tuple(channel: List[Tuple], key_index: Optional[int] = None) -> List[Tuple]

def group_tuple(channel: List[Tuple], key_index: Optional[int] = None) -> List[Tuple]:

• channel: List[Tuple] A list of tuples to be grouped by key_index
• key_index: Optional[int] Which index of the tuple to match against - if not provided, defaults to 0

channel = [(1,'A'), (1,'B'), (2,'C'), (3, 'B'), (1,'C'), (2,'A'), (3, 'D')]
group_tuple(channel) # key_index defaults to grouping by the first element (index 0)

• List[Tuple] List of grouped tuples, one per distinct key.

​

latch_filter(channel: List[Any], predicate: Union[Callable, re.Pattern, type, None]) -> List[Any]

def latch_filter(
    channel: List[Any],
    predicate: Union[Callable, re.Pattern, type, None],
) -> List[Any]:

• channel: List[Any]
• predicate: Union[Callable, re.Pattern, type, None]

• List[Any] Filtered list according to the predicate, pattern, or type. If no predicate is provided or unrecognized, returns the original channel.

​

combine(channel_0: List[Any], channel_1: List[Any], by: Optional[int] = None) -> Union[List, Dict[str, List[Any]]]

def combine(
    channel_0: List[Any],
    channel_1: List[Any],
    by: Optional[int] = None,
) -> Union[List, Dict[str, List[Any]]]:

• channel_0: List[Any] If by is provided, all elements must be tuples of the same length
• channel_1: List[Any] If by is provided, all elements must be tuples of the same length as channel_0
• by: Optional[int] If provided, which index to group by first.

• Union[List, Dict[str, List[Any]]] The Cartesian product as a list of tuples, or a dictionary of grouped products if by is used.

c0 = ['hello', 'ciao']
c1 = [1, 2, 3]
combine(c0, c1)

​

latch.functions.secrets

​

Functions

​

get_secret()

get_secret("test-secret")

• secret_name: str — Name of the secret to retrieve.

• The secret value from the response (likely a string).

• ValueError — If the HTTP response status code is not 200; the error message is taken from the response JSON as {"error": ...}.

def get_secret(secret_name: str):

​

latch.registry.project

​

Module description

​

Classes

​

ProjectNotFoundError

• Inherits: NotFoundError

​

Project

• id: str Unique identifier.

• load(self) -> None (Re-)populate this project instance’s cache. Future calls to most getters will return immediately without making a network request. Always makes a network request.
• get_display_name(self, *, load_if_missing: bool = True) -> Optional[str] Get the display name of this project. This is an opaque string that can contain any valid Unicode data. Display names are not unique and must not be used as identifiers. If load_if_missing is True and the display name is not cached, the project is loaded first.
• list_tables(self, *, load_if_missing: bool = True) -> Optional[List[Table]] List Registry tables contained in this project. If load_if_missing is True and the table list is not cached, the project is loaded first.
• update(self, *, reload_on_commit: bool = True) -> Iterator[“ProjectUpdate”] Start an update transaction. The transaction will commit when the context manager closes unless an error occurs. No changes occur until the transaction commits. If reload_on_commit is True, the project will be reloaded after commit.
• repr(self) -> str Return a developer-friendly string representation.
• str(self) -> str Alias for repr.

def load(self) -> None:
    ...

def get_display_name(self, *, load_if_missing: bool = True) -> Optional[str]:
    ...

def list_tables(self, *, load_if_missing: bool = True) -> Optional[List[Table]]:
    ...

def update(self, *, reload_on_commit: bool = True) -> Iterator["ProjectUpdate"]:
    ...

def __repr__(self) -> str:
    ...

def __str__(self) -> str:
    ...

​

ProjectUpdate

• upsert_table(self, display_name: str) -> None Creates an upsert mutation for a table with the given display name. Not idempotent; multiple calls with the same args create multiple tables.
• _add_table_upserts_selection(self, upserts: List[_ProjectTablesUpsertData], mutations: List[l.SelectionNode]) -> None Internal helper to build GraphQL selection for upsert mutations.
• delete_table(self, id: str) -> None Creates a delete mutation for the table with the given id.
• _add_table_deletes_selection(self, deletes: List[_ProjectTablesDeleteData], mutations: List[l.SelectionNode]) -> None Internal helper to build GraphQL selection for delete mutations.
• commit(self) -> None Commit this project update transaction. All pending updates are committed in one network request; the mutation is atomic. May be called multiple times.
• clear(self) -> None Remove pending updates. Cancels any uncommitted mutations.

def upsert_table(self, display_name: str) -> None:
    ...

def _add_table_upserts_selection(
    self, upserts: List[_ProjectTablesUpsertData], mutations: List[l.SelectionNode]
) -> None:
    ...

def delete_table(self, id: str) -> None:
    ...

def _add_table_deletes_selection(
    self, deletes: List[_ProjectTablesDeleteData], mutations: List[l.SelectionNode]
) -> None:
    ...

def commit(self) -> None:
    ...

def clear(self) -> None:
    ...

​

latch.account

​

Module description

​

Classes

​

Account

• Docstring: User or team workspace. Can be used to fetch related resources. :meth:current is the typical way of getting an :class:Account. If the current request signer (CLI user or execution context) lacks permissions to fetch some information, the corresponding operations will act as if the information does not exist. Update operations will usually produce errors.
• Public attribute
  • id: str — Unique identifier.
  Description based on name and usage in code.
• Methods
  • current() Get current account. Python signature:
    Copy

    Ask AI

    @classmethod
    @cache
    def current(cls) -> Self:
    

    Description: In an execution context, this is the workspace in which the execution was run. In the CLI context (when running latch commands) this is the current setting of latch workspace, which defaults to the user’s default workspace. Returns: Current account.
  • load() (Re-)populate this account instance’s cache. Python signature:
    Copy

    Ask AI

    def load(self) -> None:
    

    Description: Future calls to most getters will return immediately without making a network request. Always makes a network request.
  • *list_registry_projects(self, , load_if_missing: bool = True) -> Optional[List[Project]] Python signature:
    Copy

    Ask AI

    def list_registry_projects(
        self, *, load_if_missing: bool = True
    ) -> Optional[List[Project]]:
    

    Description: List Registry projects owned by this workspace. Parameters:
    • load_if_missing: bool — If true, :meth:load the project list if not in cache. If false, return None if not in cache.
    Returns: Projects owned by this workspace.
  • *update(self, , reload_on_commit: bool = True) -> Iterator[“AccountUpdate”] Python signature:
    Copy

    Ask AI

    @contextmanager
    def update(self, *, reload_on_commit: bool = True) -> Iterator["AccountUpdate"]:
    

    Description: Start an update transaction. The transaction will commit when the context manager closes unless an error occurs. No changes will occur until the transaction commits. The transaction can be cancelled by running :meth:AccountUpdate.clear before closing the context manager.
  • repr(self) Python signature:
    Copy

    Ask AI

    def __repr__(self):
    

    Description: Returns a string representation of the account.
  • str(self) Python signature:
    Copy

    Ask AI

    def __str__(self):
    

    Description: String representation of the account (usually same as repr).

​

AccountUpdate

• Public class description (inferred/purpose): Class for account update transactions.
• Methods
  • upsert_registry_project(self, display_name: str) Python signature:
    Copy

    Ask AI

    def upsert_registry_project(self, display_name: str):
    

    Description: Upsert a registry project. Not idempotent. Two calls with the same args will create two projects. Parameters:
    • display_name: str — Display name of the new project.
  • _add_registry_projects_upsert_selection(self, upserts: List[_AccountRegistryProjectsUpsertData], mutations: List[l.SelectionNode]) Python signature:
    Copy

    Ask AI

    def _add_registry_projects_upsert_selection(
        self,
        upserts: List[_AccountRegistryProjectsUpsertData],
        mutations: List[l.SelectionNode],
    ):
    

    Description: Prepare GraphQL selection for upsert mutations based on pending upserts.
  • delete_registry_project(self, id: str) Python signature:
    Copy

    Ask AI

    def delete_registry_project(self, id: str):
    

    Description: Delete a registry project by ID.
  • _add_registry_projects_delete_selection(self, deletes: List[_AccountRegistryProjectsDeleteData], mutations: List[l.SelectionNode]) Python signature:
    Copy

    Ask AI

    def _add_registry_projects_delete_selection(
        self,
        deletes: List[_AccountRegistryProjectsDeleteData],
        mutations: List[l.SelectionNode],
    ):
    

    Description: Prepare GraphQL selection for delete mutations based on pending deletes.
  • commit(self) -> None Python signature:
    Copy

    Ask AI

    def commit(self) -> None:
    

    Description: Commit this account update transaction. All pending updates are committed with one network request. Atomic: The entire transaction either commits or fails with an exception.
  • clear(self) Python signature:
    Copy

    Ask AI

    def clear(self):
    

    Description: Remove pending updates. May be called to cancel any pending updates that have not been committed.

​

Exceptions

​

AccountNotFoundError

• Python signature:

class AccountNotFoundError(NotFoundError): ...

• Raised when the account identified by id does not exist or permissions are insufficient to access it.

​

Examples

from latch.account import Account

# Get the current account (workspace)
acc = Account.current()

# List registry projects in the current workspace
projects = acc.list_registry_projects()

# Begin an update transaction
with acc.update() as upd:
    upd.upsert_registry_project("New Project")

# Create and commit a simple update without assuming existing data
with acc.update() as upd:
    upd.upsert_registry_project("Another Project")