latch.types package#

The remote path identifies a remote location. The remote location, either a latch or s3 url, can be inspected from an object passed to the task to reveal its remote source.

It can also to deposit the file to a latch path when the directory is returned from a task.

@task def task(dir: LatchFile):

# Manipulate directory contents locally and return back to # LatchData

return LatchDir(“./foo”, “latch:///foo”)

iterdir() → List[LatchFile | LatchDir][source]#

size_recursive()[source]#

property local_path: str#: File path local to the environment executing the task.

property remote_path: str | None#: A url referencing in object in LatchData or s3.

latch.types.directory.LatchOutputDir#

A LatchDir tagged as the output of some workflow.

The Latch Console uses this metadata to avoid checking for existence of the file at its remote path and displaying an error. This check is normally made to avoid launching workflows with LatchDirs that point to objects that don’t exist.

alias of LatchDir[LatchDir]

class latch.types.directory.LatchDirPathTransformer[source]#

Bases: FlyteDirToMultipartBlobTransformer

to_python_value(ctx: FlyteContext, lv: Literal, expected_python_type: Type[LatchDir] | PathLike) → FlyteDirectory[source]#: Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised :param ctx: FlyteContext :param lv: The received literal Value :param expected_python_type: Expected native python type that should be returned

latch.types.file module#

class latch.types.file.LatchFile(path: str | PathLike, remote_path: str | PathLike | None = None, **kwargs)[source]#

Bases: FlyteFile

Represents a file object in the context of a task execution.

The local path identifies the file object’s location on local disk in the context of a task execution. LatchFile inherits implementation of __fsopen__ from FlyteFile, so methods like open can retrieve a string representation of self.

The remote path identifies a remote location. The remote location, either a latch or s3 url, can be inspected from an object passed to the task to reveal its remote source.

It can also to deposit the file to a latch path when the object is returned from a task.

@task def task(file: LatchFile):

path = file.remote_path # inspect remote location

# Returning a different file to LatchData. return LatchFile(“./foobar.txt”, “latch:///foobar.txt”)

size()[source]#

property local_path: str#: File path local to the environment executing the task.

property remote_path: str | None#: A url referencing in object in LatchData or s3.

latch.types.file.LatchOutputFile#

A LatchFile tagged as the output of some workflow.

alias of LatchFile[LatchFile]

class latch.types.file.LatchFilePathTransformer[source]#

Bases: FlyteFilePathTransformer

to_python_value(ctx: FlyteContext, lv: Literal, expected_python_type: Type[LatchFile] | PathLike) → LatchFile[source]#: Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised :param ctx: FlyteContext :param lv: The received literal Value :param expected_python_type: Expected native python type that should be returned

latch.types.glob module#

latch.types.glob.file_glob(pattern: str, remote_directory: str, target_dir: Path | None = None) → List[LatchFile][source]#

Constructs a list of LatchFiles from a glob pattern.

Convenient utility for passing collections of files between tasks. See [nextflow’s channels](https://www.nextflow.io/docs/latest/channel.html) or [snakemake’s wildcards](https://snakemake.readthedocs.io/en/stable/snakefiles/rules.html#wildcards). for similar functionality in other orchestration tools.

The remote location of each constructed LatchFile will be consructed by appending the file name returned by the pattern to the directory represented by the remote_directory.

Parameters:

pattern – A glob pattern to match a set of files, eg. ‘*.py’. Will resolve paths with respect to the working directory of the caller.
remote_directory – A valid latch URL pointing to a directory, eg. latch:///foo. This _must_ be a directory and not a file.
target_dir – An optional Path object to define an alternate working directory for path resolution

Returns:

A list of instantiated LatchFile objects.

Intended Use:

@small_task
def task():

    ...

    return file_glob("*.fastq.gz", "latch:///fastqc_outputs")

latch.types.json module#

latch.types.json.JsonArray#

JSON-compatible list

alias of List[JsonValue]

latch.types.json.JsonObject#

JSON-compatible dictionary

alias of Dict[str, JsonValue]

latch.types.json.JsonValue#

JSON-compatible value

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

alias of Optional[Union[Dict[str, JsonValue], List[JsonValue], str, int, float, bool]]

latch.types.metadata module#

class latch.types.metadata.LatchRule(regex: str, message: str)[source]#

Bases: object

Class describing a rule that a parameter input must follow

regex: str#: A string regular expression which inputs must match

message: str#: The message to render when an input does not match the regex

property dict#

class latch.types.metadata.LatchAppearanceEnum(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]#

Bases: Enum

line = 'line'#

paragraph = 'paragraph'#

class latch.types.metadata.MultiselectOption(name: str, value: object)[source]#

Bases: object

name: str#

value: object#

class latch.types.metadata.Multiselect(options: List[latch.types.metadata.MultiselectOption] = <factory>, allow_custom: bool = False)[source]#

Bases: object

options: List[MultiselectOption]#

allow_custom: bool = False#

latch.types.metadata.LatchAppearanceType#: alias of LatchAppearanceEnum

class latch.types.metadata.LatchAuthor(name: str | None = None, email: str | None = None, github: str | None = None)[source]#

Bases: object

Class describing metadata about the workflow author

name: str | None = None#: The name of the author

email: str | None = None#: The email of the author

github: str | None = None#: A link to the github profile of the author

class latch.types.metadata.FlowBase[source]#

Bases: object

Parent class for all flow elements

Available flow elements:

Params
Text
Title
Section
Spoiler
Fork

class latch.types.metadata.Section(section: str, *flow: FlowBase)[source]#

Bases: FlowBase

Flow element that displays a child flow in a card with a given title

Example:

The LatchMetadata for the example above can be defined as follows:

from latch.types import LatchMetadata, LatchParameter
from latch.types.metadata import FlowBase, Section, Text, Params, Fork, Spoiler
from latch import workflow

flow = [
    Section(
        "Samples",
        Text(
            "Sample provided has to include an identifier for the sample (Sample name)"
            " and one or two files corresponding to the reads (single-end or paired-end, respectively)"
        ),
        Fork(
            "sample_fork",
            "Choose read type",
            paired_end=ForkBranch("Paired-end", Params("paired_end")),
            single_end=ForkBranch("Single-end", Params("single_end")),
        ),
    ),
    Section(
        "Quality threshold",
        Text(
            "Select the quality value in which a base is qualified."
            "Quality value refers to a Phred quality score"
        ),
        Params("quality_threshold"),
    ),
    Spoiler(
        "Output directory",
        Text("Name of the output directory to send results to."),
        Params("output_directory"),
    ),
]

metadata = LatchMetadata(
    display_name="fastp - Flow Tutorial",
    author=LatchAuthor(
        name="LatchBio",
    ),
    parameters={
        "sample_fork": LatchParameter(),
        "paired_end": LatchParameter(
            display_name="Paired-end reads",
            description="FASTQ files",
            batch_table_column=True,
        ),
        "single_end": LatchParameter(
            display_name="Single-end reads",
            description="FASTQ files",
            batch_table_column=True,
        ),
        "output_directory": LatchParameter(
            display_name="Output directory",
        ),
    },
    flow=flow,
)

@workflow(metadata)
def fastp(
    sample_fork: str,
    paired_end: PairedEnd,
    single_end: Optional[SingleEnd] = None,
    output_directory: str = "fastp_results",
) -> LatchDir:
    ...

section: str#: Title of the section

flow: List[FlowBase]#: Flow displayed in the section card

class latch.types.metadata.Text(text: str)[source]#

Bases: FlowBase

Flow element that displays a markdown string

text: str#: Markdown body text

class latch.types.metadata.Title(title: str)[source]#

Bases: FlowBase

Flow element that displays a markdown title

title: str#: Markdown title text

class latch.types.metadata.Params(*args: str)[source]#

Bases: FlowBase

Flow element that displays parameter widgets

params: List[str]#: Names of parameters whose widgets will be displayed. Order is preserved. Duplicates are allowed

class latch.types.metadata.Spoiler(spoiler: str, *flow: FlowBase)[source]#

Bases: FlowBase

Flow element that displays a collapsible card with a given title

spoiler: str#: Title of the spoiler

flow: List[FlowBase]#: Flow displayed in the spoiler card

class latch.types.metadata.ForkBranch(display_name: str, *flow: FlowBase)[source]#

Bases: object

Definition of a Fork branch

display_name: str#: String displayed in the fork’s multibutton

flow: List[FlowBase]#: Child flow displayed in the fork card when the branch is active

class latch.types.metadata.Fork(fork: str, display_name: str, **flows: ForkBranch)[source]#

Bases: FlowBase

Flow element that displays a set of mutually exclusive alternatives

Displays a title, followed by a horizontal multibutton for selecting a branch, then a card for the active branch

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. Order determines the order of options in the multibutton

class latch.types.metadata.LatchParameter(display_name: str | None = None, description: str | None = None, hidden: bool = False, section_title: str | None = None, placeholder: str | None = None, comment: str | None = None, output: bool = False, batch_table_column: bool = False, allow_dir: bool = True, allow_file: bool = True, appearance_type: ~latch.types.metadata.LatchAppearanceEnum | ~latch.types.metadata.Multiselect = LatchAppearanceEnum.line, rules: ~typing.List[~latch.types.metadata.LatchRule] = <factory>, detail: str | None = None, samplesheet: bool | None = None, _custom_ingestion: str | None = None)[source]#

Bases: object

Class for organizing parameter metadata

display_name: str | None = None#: The name used to display the parameter on Latch Console

description: str | None = None#: The description of the parameter’s role in the workflow

hidden: bool = False#: Whether or not the parameter should be hidden by default

section_title: str | None = None#: Whether this parameter should start a new section

placeholder: str | None = None#: What should be rendered as a placeholder in the input box of the parameter before any value is inputed.

comment: str | None = None#: Any comment on the parameter itself

output: bool = False#: Whether or not this parameter is an output (used to disable path validation before launching a workflow)

batch_table_column: bool = False#: Whether this parameter should be given a column in the batch table at the top of the workflow inputs

allow_dir: bool = True#: Whether or not this parameter should accept directories in UI

allow_file: bool = True#: Whether or not this parameter should accept files in UI.

appearance_type: LatchAppearanceEnum | Multiselect = 'line'#: Whether the parameter should be rendered as a line or paragraph (must be exactly one of either LatchAppearanceType.line or LatchAppearanceType.paragraph)

rules: List[LatchRule]#: A list of LatchRule objects that inputs to this parameter must follow

detail: str | None = None#

samplesheet: bool | None = None#: Use samplesheet input UI. Allows importing from Latch Registry. Parameter type must be a list of dataclasses

property dict#

class latch.types.metadata.SnakemakeParameter(display_name: Optional[str] = None, description: Optional[str] = None, hidden: bool = False, section_title: Optional[str] = None, placeholder: Optional[str] = None, comment: Optional[str] = None, output: bool = False, batch_table_column: bool = False, allow_dir: bool = True, allow_file: bool = True, appearance_type: Union[latch.types.metadata.LatchAppearanceEnum, latch.types.metadata.Multiselect] = <LatchAppearanceEnum.line: 'line'>, rules: List[latch.types.metadata.LatchRule] = <factory>, detail: Optional[str] = None, samplesheet: Optional[bool] = None, _custom_ingestion: Optional[str] = None, type: Union[Type[NoneType], Type[int], Type[float], Type[str], Type[bool], Type[enum.Enum], Type[latch.types.metadata._IsDataclass], Type[List[ForwardRef('ParameterType')]], Type[latch.types.file.LatchFile], Type[latch.types.directory.LatchDir], NoneType] = None, default: Optional[Any] = None)[source]#

Bases: LatchParameter

type: Type[None] | Type[int] | Type[float] | Type[str] | Type[bool] | Type[Enum] | Type[_IsDataclass] | Type[List[Type[None] | Type[int] | Type[float] | Type[str] | Type[bool] | Type[Enum] | Type[_IsDataclass] | Type[List[ParameterType]] | Type[LatchFile] | Type[LatchDir]]] | Type[LatchFile] | Type[LatchDir] | None = None#: The python type of the parameter.

default: Any | None = None#

class latch.types.metadata.SnakemakeFileParameter(display_name: str | None = None, description: str | None = None, hidden: bool = False, section_title: str | None = None, placeholder: str | None = None, comment: str | None = None, output: bool = False, batch_table_column: bool = False, allow_dir: bool = True, allow_file: bool = True, appearance_type: ~latch.types.metadata.LatchAppearanceEnum | ~latch.types.metadata.Multiselect = LatchAppearanceEnum.line, rules: ~typing.List[~latch.types.metadata.LatchRule] = <factory>, detail: str | None = None, samplesheet: bool | None = None, _custom_ingestion: str | None = None, type: ~typing.Type[~latch.types.file.LatchFile] | ~typing.Type[~latch.types.directory.LatchDir] | None = None, default: ~typing.Any | None = None, path: ~pathlib.Path | None = None, config: bool = False, download: bool = False)[source]#

Bases: SnakemakeParameter

Deprecated: use file_metadata keyword in SnakemakeMetadata instead

type: Type[LatchFile] | Type[LatchDir] | None = None#: The python type of the parameter.

path: Path | None = None#: The path where the file passed to this parameter will be copied.

config: bool = False#: Whether or not the file path is exposed in the Snakemake config

download: bool = False#: Whether or not the file is downloaded in the JIT step

class latch.types.metadata.SnakemakeFileMetadata(path: pathlib.Path, config: bool = False, download: bool = False)[source]#

Bases: object

path: Path#: The local path where the file passed to this parameter will be copied

config: bool = False#: If True, expose the file in the Snakemake config

download: bool = False#: If True, download the file in the JIT step

class latch.types.metadata.LatchMetadata(display_name: str, author: ~latch.types.metadata.LatchAuthor, documentation: str | None = None, repository: str | None = None, license: str = 'MIT', parameters: ~typing.Dict[str, ~latch.types.metadata.LatchParameter] = <factory>, wiki_url: str | None = None, video_tutorial: str | None = None, tags: ~typing.List[str] = <factory>, flow: ~typing.List[~latch.types.metadata.FlowBase] = <factory>, no_standard_bulk_execution: bool = False, _non_standard: ~typing.Dict[str, object] = <factory>)[source]#

Bases: object

Class for organizing workflow metadata

Example:

from latch.types import LatchMetadata, LatchAuthor, LatchRule, LatchAppearanceType

metadata = LatchMetadata(
    parameters={
        "read1": LatchParameter(
            display_name="Read 1",
            description="Paired-end read 1 file to be assembled.",
            hidden=True,
            section_title="Sample Reads",
            placeholder="Select a file",
            comment="This is a comment",
            output=False,
            appearance_type=LatchAppearanceType.paragraph,
            rules=[
                LatchRule(
                    regex="(.fasta|.fa|.faa|.fas)$",
                    message="Only .fasta, .fa, .fas, or .faa extensions are valid"
                )
            ],
            batch_table_column=True,  # Show this parameter in batched mode.
            # The below parameters will be displayed on the side bar of the workflow
            documentation="https://github.com/author/my_workflow/README.md",
            author=LatchAuthor(
                name="Workflow Author",
                email="licensing@company.com",
                github="https://github.com/author",
            ),
            repository="https://github.com/author/my_workflow",
            license="MIT",
            # If the workflow is public, display it under the defined categories on Latch to be more easily discovered by users
            tags=["NGS", "MAG"],
        ),
)

@workflow(metadata)
def wf(read1: LatchFile):
    ...

display_name: str#: The human-readable name of the workflow

author: LatchAuthor#: A LatchAuthor object that describes the author of the workflow

documentation: str | None = None#: A link to documentation for the workflow itself

repository: str | None = None#: A link to the repository where the code for the workflow is hosted

license: str = 'MIT'#: A SPDX identifier

parameters: Dict[str, LatchParameter]#: A dictionary mapping parameter names (strings) to LatchParameter objects

wiki_url: str | None = None#

video_tutorial: str | None = None#

tags: List[str]#

flow: List[FlowBase]#

no_standard_bulk_execution: bool = False#: Disable the standard CSV-based bulk execution. Intended for workflows that support an alternative way of processing bulk data e.g. using a samplesheet parameter

property dict#

class latch.types.metadata.DockerMetadata(username: str, secret_name: str)[source]#

Bases: object

Class describing credentials for private docker repositories

username: str#: The account username for the private repository

secret_name: str#: The name of the Latch Secret that contains the password for the private repository

class latch.types.metadata.EnvironmentConfig(use_conda: bool = False, use_container: bool = False, container_args: ~typing.List[str] = <factory>)[source]#

Bases: object

Class describing environment for spawning Snakemake tasks

use_conda: bool = False#: Use Snakemake conda directive to spawn tasks in conda environments

use_container: bool = False#: Use Snakemake container directive to spawn tasks in Docker containers

container_args: List[str]#: Additional arguments to use when running Docker containers

class latch.types.metadata.SnakemakeMetadata(display_name: str, author: ~latch.types.metadata.LatchAuthor, documentation: str | None = None, repository: str | None = None, license: str = 'MIT', parameters: ~typing.Dict[str, ~latch.types.metadata.SnakemakeParameter] = <factory>, wiki_url: str | None = None, video_tutorial: str | None = None, tags: ~typing.List[str] = <factory>, flow: ~typing.List[~latch.types.metadata.FlowBase] = <factory>, no_standard_bulk_execution: bool = False, _non_standard: ~typing.Dict[str, object] = <factory>, output_dir: ~latch.types.directory.LatchDir | None = None, name: str | None = None, docker_metadata: ~latch.types.metadata.DockerMetadata | None = None, env_config: ~latch.types.metadata.EnvironmentConfig = <factory>, file_metadata: ~typing.Dict[str, ~latch.types.metadata.SnakemakeFileMetadata | ~typing.Dict[str, ~latch.types.metadata.SnakemakeFileMetadata | FileMetadata]] = <factory>, cores: int = 4)[source]#

Bases: LatchMetadata

Class for organizing Snakemake workflow metadata

output_dir: LatchDir | None = None#: Directory for snakemake workflow outputs

name: str | None = None#: Name of the workflow

docker_metadata: DockerMetadata | None = None#: Credentials configuration for private docker repositories

env_config: EnvironmentConfig#: Environment configuration for spawning Snakemake tasks

parameters: Dict[str, SnakemakeParameter]#: A dictionary mapping parameter names (strings) to SnakemakeParameter objects

file_metadata: Dict[str, SnakemakeFileMetadata | Dict[str, SnakemakeFileMetadata | FileMetadata]]#: A dictionary mapping parameter names to SnakemakeFileMetadata objects

cores: int = 4#: Number of cores to use for Snakemake tasks (equivalent of Snakemake’s –cores flag)

validate()[source]#

latch.types.utils module#

latch.types.utils.is_valid_url(raw_url: str | Path) → bool[source]#: A valid URL (as a source or destination of a LatchFile) must: * contain a latch or s3 scheme * contain an absolute path

latch.types.utils.format_path(path: str) → str[source]#

latch.types package#

Submodules#

latch.types.directory module#

latch.types.file module#

latch.types.glob module#

latch.types.json module#

latch.types.metadata module#

latch.types.utils module#

Module contents#