Latch SDK API Reference

​

Table of Contents

• Core SDK
  • latch.init - Main SDK initialization and imports
• Workflow Resources
  • latch.resources.tasks - Task decorators and execution
  • latch.resources.workflow - Workflow definition and execution
  • latch.resources.conditional - Conditional workflow logic
  • latch.resources.map_tasks - Parallel task execution
  • latch.resources.reference_workflow - Workflow references
• Data Types
  • latch.types.init - Type system overview
  • latch.types.file - File handling (LatchFile, LatchOutputFile)
  • latch.types.directory - Directory handling (LatchDir, LatchOutputDir)
  • latch.types.metadata - Workflow metadata and UI configuration
  • latch.types.glob - File pattern matching (file_glob)
• Utility Functions
  • latch.functions.messages - Console messaging (message)
  • latch.functions.operators - Data manipulation operators (deprecated)
  • latch.functions.secrets - Secret management (get_secret)
• Data Management
  • latch.ldata - Latch Data cloud storage (LPath)

​

latch.init

​

latch.resources.tasks

from latch.resources.tasks import medium_task

@medium_task
def my_task(a: int) -> str:
  ...

​

Functions

​

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

• A partial Flyte task configured with the specified Pod settings

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

from latch.resources.tasks import custom_memory_optimized_task

@custom_memory_optimized_task(cpu=16, memory=128)
def my_task(a: int) -> str:
  ...

​

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

• A partial Flyte task configured for either a dynamic or a static custom task

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

from latch.resources.tasks import custom_task

@custom_task(cpu=8, memory=32, storage_gib=200)
def my_task(a: int) -> str:
  ...

​

lustre_setup_task()

• None

• A partial Flyte task

def lustre_setup_task():

from latch.resources.tasks import lustre_setup_task

@lustre_setup_task()
def setup_workdir(pvc_name: str) -> str:
    os.chmod("/nf-workdir", 0o777)
    return pvc_name

​

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

• cpu (int): CPU cores
• memory (int): Memory in GiB
• storage_gib (int, default 50): Storage in GiB

• A partial Flyte task

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

# Full code example: https://github.com/latchbio-nfcore/atacseq/blob/370c45694bca54a89a7cc65902b0fb54747a87b5/wf/entrypoint.py#L98

from latch.resources.tasks import nextflow_runtime_task

@nextflow_runtime_task(cpu=4, memory=16, storage_gib=64)
def nextflow_runtime(
    pvc_name: str,
    input: LatchFile,
    outdir: LatchDir,
    fasta: LatchFile,
) -> None:
    ...

​

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

• None

• A partial Flyte task for each respective instance type and resources

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

from latch.resources.tasks import g6e_xlarge_task

@g6e_xlarge_task
def my_task(a: int) -> str:
  ...

​

v100_x1_task, v100_x4_task, v100_x8_task

• None

• A partial Flyte task for each respective instance type and resources

v100_x1_task = functools.partial(
    task,
    task_config=_get_v100_pod("v100-x1", cpu=4, memory_gib=48, gpus=1)
)

from latch.resources.tasks import v100_x1_task

@v100_x1_task
def my_task(a: int) -> str:
  ...

​

latch.resources.workflow

​

Functions

​

workflow()

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

• metadata (Union[LatchMetadata, Callable]): Either a LatchMetadata instance or the function to decorate (when used without parentheses)

• Union[PythonFunctionWorkflow, Callable]: A PythonFunctionWorkflow if used directly or a decorator if used with arguments

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.directory import LatchDir, LatchOutputDir
from latch.types.metadata import LatchAuthor, LatchMetadata, LatchParameter

metadata = LatchMetadata(
    display_name="Target Workflow",
    author=LatchAuthor(
        name="Your Name",
    ),
    parameters={
        "input_directory": LatchParameter(
            display_name="Input Directory",
            batch_table_column=True,  # Show this parameter in batched mode.
        ),
        "output_directory": LatchParameter(
            display_name="Output Directory",
            batch_table_column=True,  # Show this parameter in batched mode.
        ),
    },
)


@workflow(metadata)
def template_workflow(
    input_directory: LatchDir, output_directory: LatchOutputDir
) -> LatchOutputDir:
    return task(input_directory=input_directory, output_directory=output_directory)

​

nextflow_workflow()

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

• metadata (NextflowMetadata): Metadata for the Nextflow-style workflow

• Callable[[Callable], PythonFunctionWorkflow]: A decorator compatible with workflow

from latch.resources.workflow import nextflow_workflow
from latch.types.metadata import NextflowMetadata, NextflowParameter
from latch.types.file import LatchFile
from latch.types.directory import LatchDir
from latch.types.metadata import LatchAuthor, NextflowRuntimeResources

generated_parameters = {
    "input": NextflowParameter(
        type=LatchFile,
        display_name="Input",
        description="Input file description",
    ),
    ...
}

nextflow_metadata = NextflowMetadata(
    display_name="nf-core/atacseq",
    author=LatchAuthor(
        name="nf-core",
    ),
    repository="https://github.com/latchbio-nfcore/atacseq",
    parameters=generated_parameters,
    runtime_resources=NextflowRuntimeResources(
        cpus=4,
        memory=8,
        storage_gib=100,
    ),
    log_dir=LatchDir("latch:///your_log_dir"),
)

@nextflow_workflow(nextflow_metadata)
def nf_nf_core_atacseq(...) -> LatchOutputDir:
    ...
    return LatchOutputDir("latch:///your_output_dir")

​

latch.resources.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:
    ...

from latch.resources.tasks import small_task
from latch import create_conditional_section

@small_task
def square(n: float) -> float:
    """
    **Parameters:**
        - `n` (float): Name of the parameter for the task is derived from the name of the input variable, and the type is automatically mapped to Types.Integer
    **Returns:**
        - `float`: The label for the output is automatically assigned and the type is deduced from the annotation
    """
    return n * n


@small_task
def double(n: float) -> float:
    """
    **Parameters:**
        - `n` (float): Name of the parameter for the task is derived from the name of the input variable and the type is mapped to `Types.Integer`
    **Returns:**
        - `float`: The label for the output is auto-assigned and the type is deduced from the annotation
    """
    return 2 * n

@workflow
def multiplier(my_input: float) -> float:
    result_1 = double(n=my_input)
    result_2 =  (
        create_conditional_section("fractions")
        .if_((result_1 < 0.0)).then(double(n=result_1))
        .elif_((result_1 > 0.0)).then(square(n=result_1))
        .else_().fail("Only nonzero values allowed")
    )
    result_3 = double(n=result_2)
    return result_3

​

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

from latch.resources.tasks import small_task, map_task
from latch.resources.workflow import workflow

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

@small_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)
    coalesced = coalesce(b=mapped_out)
    return coalesced

​

latch.resources.reference_workflow

​

Functions

​

workflow_reference()

• name (str): The name of the launch plan to reference
• version (str): The version of the launch plan to reference

• 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

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

​

Functions

​

file_glob()

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

• pattern (str): A glob pattern to match a set of files, e.g. '*.py'. Will resolve paths with respect to the working directory of the caller.
• 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.

• List[LatchFile]: A list of instantiated LatchFile objects.

@small_task
def process_fastq_files():
    # Get all .fastq.gz files from the current directory
    # and map them to a remote directory
    fastq_files = file_glob("*.fastq.gz", "latch:///fastqc_outputs")
    
    for fastq_file in fastq_files:
        # Process each file
        process_file(fastq_file)
    
    return fastq_files

​

Classes

• LatchDir - See latch.types.directory
• LatchOutputDir - See latch.types.directory
• LatchFile - See latch.types.file
• LatchOutputFile - See latch.types.file
• DockerMetadata - See latch.types.metadata
• Fork - See latch.types.metadata
• ForkBranch - See latch.types.metadata
• LatchAppearanceType - See latch.types.metadata
• LatchAuthor - See latch.types.metadata
• LatchMetadata - See latch.types.metadata
• LatchParameter - See latch.types.metadata
• LatchRule - See latch.types.metadata
• Params - See latch.types.metadata
• Section - See latch.types.metadata
• Spoiler - See latch.types.metadata
• Text - See latch.types.metadata

​

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

• size(self) -> int Returns the size of the remote data via LPath(self.remote_path).size().
• 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).

# 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")

​

latch.types.directory

​

Classes

​

LatchDir

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

• path (Union[str, PathLike]): The local path to the directory
• remote_path (Optional[PathLike]): The remote path (latch:// or s3:// URL) where the directory is stored

• local_path (str): Local directory path for the environment executing the task
• remote_path (Optional[str]): Remote URL referencing the directory (LatchData or S3)

• iterdir() -> List[Union[LatchFile, LatchDir]]: Returns a list of the directory’s children (files and subdirectories)
• size_recursive() -> int: Returns the total size of the directory and all its contents recursively

from latch.types.directory import LatchDir
from pathlib import Path

# Create a local directory
local_dir = LatchDir("./my_directory")

# Create a directory with remote path
remote_dir = LatchDir("./my_directory", "latch:///remote_directory")

# In a workflow task
@small_task
def process_directory(input_dir: LatchDir) -> LatchDir:
    # Access local path
    local_path = Path(input_dir.local_path)
    
    # List directory contents
    for item in input_dir.iterdir():
        if isinstance(item, LatchFile):
            print(f"Found file: {item}")
        elif isinstance(item, LatchDir):
            print(f"Found subdirectory: {item}")
    
    # Get directory size
    total_size = input_dir.size_recursive()
    print(f"Directory size: {total_size} bytes")
    
    # Create output directory
    return LatchDir("./output", "latch:///output_directory")

# In a Jupyter notebook or standalone script
def explore_remote_directory():
    # Create a LatchDir pointing to a remote directory
    remote_dir = LatchDir("latch:///my_data")
    
    # List contents of the remote directory
    for item in remote_dir.iterdir():
        print(f"Item: {item}")
        if hasattr(item, 'remote_path'):
            print(f"  Remote path: {item.remote_path}")
    
    # Get directory size
    total_size = remote_dir.size_recursive()
    print(f"Total directory size: {total_size} bytes")

​

LatchOutputDir

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

from latch.types.directory import LatchDir, LatchOutputDir

@small_task
def create_output() -> LatchOutputDir:
    return LatchDir("./results", "latch:///my_workflow_output")

​

latch.types.metadata

​

Functions

​

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

• samples (List[DC]): List of dataclass instances to convert to CSV
• t (DC): The dataclass type to use for column headers
• delim (str): CSV delimiter, defaults to ”,”

• Path: Path to the created samplesheet.csv file

from dataclasses import dataclass
from latch.types.metadata import default_samplesheet_constructor

@dataclass
class SampleData:
    sample_id: str
    condition: str
    replicate: int

# Create sample data
samples = [
    SampleData("sample1", "control", 1),
    SampleData("sample2", "treatment", 1),
    SampleData("sample3", "control", 2),
]

# Generate samplesheet
csv_path = default_samplesheet_constructor(samples, t=SampleData)
print(f"Created samplesheet at: {csv_path}")

​

Classes

​

LatchRule

• regex (str): Regular expression pattern that inputs must match
• message (str): Error message displayed when validation fails

from latch.types.metadata import LatchRule

# Validate email format
email_rule = LatchRule(
    regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
    message="Please enter a valid email address"
)

# Validate file extension
fastq_rule = LatchRule(
    regex=r"\.(fastq|fq)(\.gz)?$",
    message="File must be a FASTQ file (.fastq, .fq, .fastq.gz, or .fq.gz)"
)

​

LatchAppearanceEnum

• line: Single-line text input
• paragraph: Multi-line text area

from latch.types.metadata import LatchAppearanceEnum

# Single line input for short text
short_input = LatchAppearanceEnum.line

# Multi-line input for longer text
long_input = LatchAppearanceEnum.paragraph

​

MultiselectOption

• name (str): Display name shown in the UI
• value (object): Value associated with this option

​

Multiselect

• options (List[MultiselectOption]): List of available options
• allow_custom (bool): Whether users can enter custom values

from latch.types.metadata import Multiselect, MultiselectOption

# Create multiselect for organism selection
organism_multiselect = Multiselect(
    options=[
        MultiselectOption("Human", "homo_sapiens"),
        MultiselectOption("Mouse", "mus_musculus"),
        MultiselectOption("Drosophila", "drosophila_melanogaster"),
    ],
    allow_custom=True  # Allow users to enter custom organism names
)

​

LatchAuthor

• name (Optional[str]): Author’s name
• email (Optional[str]): Author’s email address
• github (Optional[str]): Link to author’s GitHub profile

from latch.types.metadata import LatchAuthor

author = LatchAuthor(
    name="Dr. Jane Smith",
    email="jane.smith@university.edu",
    github="https://github.com/janesmith"
)

​

UI Flow Elements

​

FlowBase

​

Section

Section(section: str, *flow: FlowBase)

• section (str): Title displayed on the section card
• *flow (FlowBase): Variable number of flow elements to display in the section

​

Text

• text (str): Markdown content to display

​

Title

• title (str): Markdown title text

​

Params

Params(*args: str)

• *args (str): Names of parameters to display

​

Spoiler

Spoiler(spoiler: str, *flow: FlowBase)

• spoiler (str): Title of the collapsible section
• *flow (FlowBase): Flow elements to display when expanded

from latch.types.metadata import Section, Text, Params, Spoiler

# Create a workflow UI flow
flow = [
    Text("## RNA-seq Analysis Workflow"),
    Text("This workflow performs differential gene expression analysis."),
    
    Section("Input Parameters", 
        Params("input_files", "reference_genome"),
        Text("Select your input FASTQ files and reference genome.")
    ),
    
    Section("Analysis Options",
        Params("min_reads", "p_value_threshold"),
        Spoiler("Advanced Options",
            Params("threads", "memory_limit"),
            Text("Configure advanced computational parameters.")
        )
    )
]

​

ForkBranch

ForkBranch(display_name: str, *flow: FlowBase)

• display_name (str): Text displayed on the branch button
• *flow (FlowBase): Flow elements to display when this branch is active

​

Fork

Fork(fork: str, display_name: str, **flows: ForkBranch)

• fork (str): Name of the string parameter that stores the selected branch key
• display_name (str): Title shown above the fork selector
• **flows (ForkBranch): Named branches, where keys become the parameter values

from latch.types.metadata import Fork, ForkBranch, Params, Text

# Create conditional analysis options
analysis_fork = Fork(
    fork="analysis_type",
    display_name="Analysis Type",
    differential=ForkBranch(
        "Differential Expression",
        Params("p_value_threshold", "fold_change_threshold"),
        Text("Configure parameters for differential expression analysis.")
    ),
    pathway=ForkBranch(
        "Pathway Analysis", 
        Params("pathway_database", "enrichment_method"),
        Text("Configure parameters for pathway enrichment analysis.")
    )
)

​

Core Metadata Classes

​

LatchParameter

• display_name (Optional[str]): Human-readable name for the parameter
• description (Optional[str]): Help text describing the parameter
• hidden (bool): Whether to hide the parameter by default
• placeholder (Optional[str]): Placeholder text in input fields
• output (bool): Whether this parameter represents a workflow output
• rules (List[LatchRule]): Validation rules for the parameter
• appearance_type (LatchAppearance): How to render the input (line/paragraph/multiselect)

• samplesheet (Optional[bool]): Enable samplesheet input UI
• allowed_tables (Optional[List[int]]): Registry table IDs allowed for samplesheet

from latch.types.metadata import LatchParameter, LatchRule, LatchAppearanceEnum

# Basic parameter
input_file_param = LatchParameter(
    display_name="Input FASTQ Files",
    description="Select your input FASTQ files for analysis",
    placeholder="Choose files...",
    rules=[
        LatchRule(
            regex=r"\.(fastq|fq)(\.gz)?$",
            message="Please select FASTQ files"
        )
    ]
)

# Samplesheet parameter
sample_param = LatchParameter(
    display_name="Sample Information",
    description="Upload a samplesheet with sample metadata",
    samplesheet=True,
    allowed_tables=[123, 456]  # Specific registry table IDs
)

# Text parameter with validation
email_param = LatchParameter(
    display_name="Email Address",
    description="Your email for notifications",
    appearance_type=LatchAppearanceEnum.line,
    rules=[
        LatchRule(
            regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$",
            message="Please enter a valid email address"
        )
    ]
)

​

LatchMetadata

• display_name (str): Human-readable workflow name
• author (LatchAuthor): Workflow author information
• documentation (Optional[str]): Link to workflow documentation
• repository (Optional[str]): Link to source code repository
• license (str): SPDX license identifier
• parameters (Dict[str, LatchParameter]): Parameter definitions
• flow (List[FlowBase]): UI layout configuration

• tags (List[str]): Categorization tags
• wiki_url (Optional[str]): Link to wiki documentation
• video_tutorial (Optional[str]): Link to tutorial video
• about_page_path (Optional[Path]): Path to markdown about page

from latch.types.metadata import (
    LatchMetadata, LatchAuthor, LatchParameter, 
    Section, Text, Params
)

# Define workflow metadata
metadata = LatchMetadata(
    display_name="RNA-seq Differential Expression Analysis",
    author=LatchAuthor(
        name="Dr. Jane Smith",
        email="jane.smith@university.edu",
        github="https://github.com/janesmith"
    ),
    documentation="https://docs.example.com/rnaseq-workflow",
    repository="https://github.com/janesmith/rnaseq-workflow",
    license="MIT",
    tags=["RNA-seq", "differential-expression", "bioinformatics"],
    
    # Define parameters
    parameters={
        "input_files": LatchParameter(
            display_name="Input FASTQ Files",
            description="Select your input FASTQ files"
        ),
        "reference_genome": LatchParameter(
            display_name="Reference Genome",
            description="Choose a reference genome"
        ),
        "p_value_threshold": LatchParameter(
            display_name="P-value Threshold",
            description="Statistical significance threshold",
            placeholder="0.05"
        )
    },
    
    # Define UI layout
    flow=[
        Text("## RNA-seq Analysis Workflow"),
        Text("This workflow performs differential gene expression analysis using DESeq2."),
        
        Section("Input Data",
            Params("input_files", "reference_genome"),
            Text("Select your input files and reference genome.")
        ),
        
        Section("Analysis Parameters",
            Params("p_value_threshold"),
            Text("Configure statistical parameters for the analysis.")
        )
    ]
)

​

Integration Metadata Classes

​

DockerMetadata

• username (str): Docker registry username
• secret_name (str): Name of the secret containing the password

​

SnakemakeMetadata(LatchMetadata)

• output_dir (Optional[LatchDir]): Directory for Snakemake outputs
• docker_metadata (Optional[DockerMetadata]): Docker registry credentials
• cores (int): Number of cores for Snakemake execution
• parameters (Dict[str, SnakemakeParameter]): Snakemake-specific parameter metadata

​

NextflowMetadata(LatchMetadata)

• runtime_resources (NextflowRuntimeResources): Computational resources
• execution_profiles (List[str]): Nextflow execution profiles
• log_dir (Optional[LatchDir]): Directory for Nextflow logs
• parameters (Dict[str, NextflowParameter]): Nextflow-specific parameter metadata

​

Helper Classes

​

SnakemakeParameter(Generic[T], LatchParameter)

​

NextflowParameter(Generic[T], LatchParameter)

​

NextflowRuntimeResources

• cpus (Optional[int]): Number of CPUs
• memory (Optional[int]): Memory in GiB
• storage_gib (Optional[int]): Storage in GiB
• storage_expiration_hours (int): Workdir retention time

​

latch.types.glob

​

Functions

​

file_glob()

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

• pattern (str): Glob pattern to match files (e.g., "*.fastq.gz", "data/*.csv")
• remote_directory (str): Latch URL pointing to a directory (e.g., "latch:///my_data")
• target_dir (Optional[Path]): Alternative working directory for pattern resolution

• List[LatchFile]: List of LatchFile objects with local paths matching the pattern and remote paths in the specified directory

from latch.types.glob import file_glob
from pathlib import Path

# Basic usage - match all FASTQ files in current directory
fastq_files = file_glob("*.fastq.gz", "latch:///rnaseq_inputs")

# Match files in subdirectories
all_csv_files = file_glob("data/**/*.csv", "latch:///processed_data")

# Use custom working directory
custom_files = file_glob(
    "*.bam", 
    "latch:///alignment_outputs",
    target_dir=Path("/path/to/working/dir")
)

# Process multiple file types
@small_task
def process_sequencing_files():
    # Get all sequencing files
    fastq_files = file_glob("*.fastq.gz", "latch:///raw_data")
    bam_files = file_glob("*.bam", "latch:///alignments")
    
    # Process each file
    for file in fastq_files + bam_files:
        process_file(file)
    
    return fastq_files + bam_files

• The remote_directory must be a valid Latch URL pointing to a directory (not a file)
• If the remote directory URL is invalid, an empty list is returned
• Each matched file gets a remote path constructed by appending the filename to the remote directory
• This is particularly useful for workflows that need to process multiple files of the same type

​

latch.functions.messages

​

Functions

​

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

• typ (str): Message type determining display style. Options:
  • "info": Informational messages (blue styling)
  • "warning": Warning messages (yellow/orange styling)
  • "error": Error messages (red styling)
• data (Dict[str, Any]): Message content with required keys:
  • "title" (str): Brief message title
  • "body" (str): Detailed message content

• None

• RuntimeError: If message processing fails

from latch.functions.messages import message

@small_task
def process_samples(input_file: LatchFile) -> str:
    # Info message - task progress
    message("info", {
        "title": "Processing Started",
        "body": f"Analyzing {input_file} with 1000 samples"
    })
    
    try:
        # Process the file
        result = analyze_file(input_file)
        
        # Success message
        message("info", {
            "title": "Analysis Complete",
            "body": f"Successfully processed {len(result)} samples"
        })
        
        return result
        
    except ValueError as e:
        # Error message with helpful guidance
        message("error", {
            "title": "Invalid File Format",
            "body": f"Expected CSV format with columns: sample_id, condition, replicate. Error: {str(e)}"
        })
        raise
        
    except FileNotFoundError:
        # Warning message
        message("warning", {
            "title": "File Not Found",
            "body": "Input file could not be located. Please check the file path and try again."
        })
        raise

@small_task
def validate_parameters(threshold: float) -> bool:
    if threshold < 0 or threshold > 1:
        message("error", {
            "title": "Invalid Parameter Value",
            "body": f"Threshold must be between 0 and 1, got {threshold}. Please adjust your input."
        })
        return False
    
    if threshold > 0.05:
        message("warning", {
            "title": "High Threshold Warning",
            "body": f"Threshold of {threshold} is quite high. Consider using a lower value (e.g., 0.05) for more sensitive results."
        })
    
    return True

​

latch.functions.operators

​

Functions

​

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

• left (Dict[str, Any]): Left dictionary (all keys preserved)
• right (Dict[str, Any]): Right dictionary (matched keys only)

• Dict[str, Any]: Dictionary with all left keys, combined with right values where keys match

left = {"a": 1, "b": 2, "c": 3}
right = {"b": 20, "c": 30, "d": 40}
result = left_join(left, right)
# Result: {"a": 1, "b": [2, 20], "c": [3, 30]}

​

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

• left (Dict[str, Any]): Left dictionary (matched keys only)
• right (Dict[str, Any]): Right dictionary (all keys preserved)

• Dict[str, Any]: Dictionary with all right keys, combined with left values where keys match

left = {"a": 1, "b": 2, "c": 3}
right = {"b": 20, "c": 30, "d": 40}
result = right_join(left, right)
# Result: {"b": [2, 20], "c": [3, 30], "d": 40}

​

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

• left (Dict[str, Any]): Left dictionary
• right (Dict[str, Any]): Right dictionary

• Dict[str, Any]: Dictionary containing only keys present in both inputs, with combined values

left = {"a": 1, "b": 2, "c": 3}
right = {"b": 20, "c": 30, "d": 40}
result = inner_join(left, right)
# Result: {"b": [2, 20], "c": [3, 30]}

​

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

• left (Dict[str, Any]): Left dictionary
• right (Dict[str, Any]): Right dictionary

• Dict[str, Any]: Dictionary containing all keys from both inputs, with combined values where keys match

left = {"a": 1, "b": 2, "c": 3}
right = {"b": 20, "c": 30, "d": 40}
result = outer_join(left, right)
# Result: {"a": 1, "b": [2, 20], "c": [3, 30], "d": 40}

​

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

• channel (List[Tuple]): List of tuples to group
• key_index (Optional[int]): Index to group by (defaults to 0)

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

channel = [(1, 'A'), (1, 'B'), (2, 'C'), (3, 'B'), (1, 'C'), (2, 'A'), (3, 'D')]
result = group_tuple(channel)  # Group by first element (index 0)
# Result: [(1, ['A', 'B', 'C']), (2, ['C', 'A']), (3, ['B', 'D'])]

# Group by second element (index 1)
result2 = group_tuple(channel, key_index=1)
# Result: [('A', [1, 2]), ('B', [1, 3]), ('C', [2, 1]), ('D', [3])]

​

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

• channel (List[Any]): List to filter
• predicate (Union[Callable, re.Pattern, type, None]): Filter criteria:
  • Callable: Function that returns True/False for each item
  • re.Pattern: Regex pattern to match against strings
  • type: Type to filter by (e.g., str, int)
  • None: Returns original list unchanged

• List[Any]: Filtered list based on the predicate

import re

# Filter by function
numbers = [1, 2, 3, 4, 5, 6]
evens = latch_filter(numbers, lambda x: x % 2 == 0)
# Result: [2, 4, 6]

# Filter by regex pattern
strings = ["hello", "world", "test123", "abc"]
pattern = re.compile(r'\d')
with_numbers = latch_filter(strings, pattern)
# Result: ["test123"]

# Filter by type
mixed = [1, "hello", 2.5, "world", 3]
strings_only = latch_filter(mixed, str)
# Result: ["hello", "world"]

​

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

• channel_0 (List[Any]): First list to combine
• channel_1 (List[Any]): Second list to combine
• by (Optional[int]): If provided, group tuples by this index before combining

• Union[List, Dict[str, List[Any]]]:
  • If by is None: List of tuples representing the Cartesian product
  • If by is provided: Dictionary with grouped products

# Basic Cartesian product
c0 = ['hello', 'ciao']
c1 = [1, 2, 3]
result = combine(c0, c1)
# Result: [('hello', 1), ('hello', 2), ('hello', 3), ('ciao', 1), ('ciao', 2), ('ciao', 3)]

# With grouping by tuple index
c0_tuples = [('A', 1), ('A', 2), ('B', 3)]
c1_tuples = [('A', 'x'), ('A', 'y'), ('B', 'z')]
result = combine(c0_tuples, c1_tuples, by=0)  # Group by first element
# Result: {
#   'A': [('A', 1, 'A', 'x'), ('A', 1, 'A', 'y'), ('A', 2, 'A', 'x'), ('A', 2, 'A', 'y')],
#   'B': [('B', 3, 'B', 'z')]
# }

​

latch.functions.secrets

​

Functions

​

get_secret(secret_name: str) -> str

• secret_name (str): Name of the secret to retrieve

• str: The decrypted secret value

from latch.functions.secrets import get_secret

@small_task
def process_with_api_key() -> str:
    # Retrieve API key from secrets
    api_key = get_secret("my-api-key")
    
    # Use the API key for authentication
    response = make_api_call(api_key)
    return response

​

latch.ldata

​

Classes

​

LPath

LPath(path: str)

• path (str): The Latch path, must start with “latch://”

• Lazy Loading: Metadata is fetched on-demand to minimize network requests
• Caching: Metadata is cached after first fetch for improved performance
• Path Operations: Supports path joining with / operator
• File/Directory Operations: Upload, download, copy, delete, and directory listing
• Metadata Access: Get file size, content type, version ID, and node information

• fetch_metadata(): Force refresh of all cached metadata
• node_id(load_if_missing=True): Get the unique node ID
• name(load_if_missing=True): Get the file/directory name
• type(load_if_missing=True): Get the node type (file, directory, etc.)
• size(load_if_missing=True): Get file size in bytes
• size_recursive(load_if_missing=True): Get total size including subdirectories
• content_type(load_if_missing=True): Get MIME content type
• version_id(load_if_missing=True): Get version identifier
• is_dir(load_if_missing=True): Check if path is a directory

• iterdir(): List contents of directory (non-recursive)
• mkdirp(): Create directory and all parent directories
• rmr(): Recursively delete files and directories

• upload_from(src: Path, show_progress_bar=False): Upload local file/directory
• download(dst=None, show_progress_bar=False, cache=False): Download to local path
• copy_to(dst: LPath): Copy to another LPath

• __truediv__(other): Join paths using / operator

from latch.ldata import LPath
from pathlib import Path

# Create LPath instances
data_dir = LPath("latch:///my_workspace/data")
input_file = LPath("latch:///my_workspace/inputs/sample.fastq")
output_dir = LPath("latch:///my_workspace/results")

# Check if paths exist and get metadata
if data_dir.is_dir():
    print(f"Directory size: {data_dir.size_recursive()} bytes")

# List directory contents
for item in data_dir.iterdir():
    print(f"{item.name()}: {item.type()}")

# Upload local files
local_file = Path("local_data.csv")
data_dir.upload_from(local_file)

# Download files
downloaded_path = input_file.download()
print(f"Downloaded to: {downloaded_path}")

# Copy between Latch paths
backup_file = LPath("latch:///my_workspace/backups/sample.fastq")
input_file.copy_to(backup_file)

# Create directories
new_dir = LPath("latch:///my_workspace/new_analysis")
new_dir.mkdirp()

# Path joining
results_file = output_dir / "analysis_results.txt"
print(f"Full path: {results_file.path}")

# Workflow integration
@small_task
def process_data(input_path: LPath) -> LPath:
    # Download input
    local_input = input_path.download()
    
    # Process the file
    result = analyze_file(local_input)
    
    # Upload result
    output_path = LPath("latch:///my_workspace/results/processed.txt")
    output_path.upload_from(Path(result))
    
    return output_path

# Batch operations
@small_task
def batch_upload(input_dir: Path) -> List[LPath]:
    uploaded_paths = []
    
    for file_path in input_dir.glob("*.csv"):
        # Create corresponding LPath
        latch_path = LPath(f"latch:///my_workspace/data/{file_path.name}")
        
        # Upload file
        latch_path.upload_from(file_path)
        uploaded_paths.append(latch_path)
    
    return uploaded_paths

from latch.ldata import LPath, LatchPathError

try:
    # This will raise LatchPathError if path doesn't exist
    file_path = LPath("latch:///nonexistent/file.txt")
    size = file_path.size()
    
except LatchPathError as e:
    print(f"Path error: {e}")
    print(f"Remote path: {e.remote_path}")
    print(f"Account ID: {e.acc_id}")

• Use load_if_missing=False when you know metadata is already cached
• Call fetch_metadata() to refresh stale cache when needed
• Use cache=True in download() for repeated downloads of the same file
• Handle LatchPathError for robust error handling
• Use path joining with / operator for cleaner code
• Check is_dir() before calling directory-specific methods

​

LatchPathError

• message (str): Error description
• remote_path (Optional[str]): The Latch path that caused the error
• acc_id (Optional[str]): Account ID associated with the error

​

LDataNodeType

• account_root: Root directory of an account
• dir: Regular directory
• obj: File object
• mount: Mounted storage
• link: Symbolic link
• mount_gcp: Google Cloud Platform mount
• mount_azure: Azure mount

from latch.ldata import LPath, LDataNodeType

path = LPath("latch:///my_workspace/data")
node_type = path.type()

if node_type == LDataNodeType.dir:
    print("This is a directory")
elif node_type == LDataNodeType.obj:
    print("This is a file")