latch.types package#

Submodules#

latch.types.directory module#

class latch.types.directory.LatchDir(path: Union[str, os.PathLike], remote_path: Optional[os.PathLike] = None, **kwargs)[source]#

Bases: flytekit.types.directory.types.FlyteDirectory

Represents a directory in the context of a task execution.

The local path identifies the directory location on local disk in the context of a task execution.

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”)

property local_path: str#

File path local to the environment executing the task.

property remote_path: Optional[str]#

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 latch.types.directory.LatchDir[latch.types.directory.LatchDir]

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

Bases: flytekit.types.directory.types.FlyteDirToMultipartBlobTransformer

to_python_value(ctx: flytekit.core.context_manager.FlyteContext, lv: flytekit.models.literals.Literal, expected_python_type: Union[Type[latch.types.directory.LatchDir], os.PathLike]) flytekit.types.directory.types.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: Union[str, os.PathLike], remote_path: Optional[Union[str, os.PathLike]] = None, **kwargs)[source]#

Bases: flytekit.types.file.file.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”)

property local_path: str#

File path local to the environment executing the task.

property remote_path: Optional[str]#

A url referencing in object in LatchData or s3.

latch.types.file.LatchOutputFile#

A LatchFile 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 LatchFiles that point to objects that don’t exist.

alias of latch.types.file.LatchFile[latch.types.file.LatchFile]

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

Bases: flytekit.types.file.file.FlyteFilePathTransformer

to_python_value(ctx: flytekit.core.context_manager.FlyteContext, lv: flytekit.models.literals.Literal, expected_python_type: Union[Type[latch.types.file.LatchFile], os.PathLike]) flytekit.types.file.file.FlyteFile[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: Optional[pathlib.Path] = None) List[latch.types.file.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.LatchAppearanceType(value)[source]#

Bases: enum.Enum

An enumeration.

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

Bases: object

Class describing metadata about the workflow author

name: Optional[str] = None#

The name of the author

email: Optional[str] = None#

The email of the author

github: Optional[str] = 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:

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

Bases: latch.types.metadata.FlowBase

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

Example:

![Example of a user interface for a workflow with a custom flow](../assets/flow-example/flow_example_1.png)

![Example of a spoiler flow element](../assets/flow-example/flow_example_spoiler.png)

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

```python 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[latch.types.metadata.FlowBase]#

Flow displayed in the section card

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

Bases: latch.types.metadata.FlowBase

Flow element that displays a markdown string

text: str#

Markdown body text

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

Bases: latch.types.metadata.FlowBase

Flow element that displays a markdown title

title: str#

Markdown title text

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

Bases: latch.types.metadata.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: latch.types.metadata.FlowBase)[source]#

Bases: latch.types.metadata.FlowBase

Flow element that displays a collapsible card with a given title

spoiler: str#

Title of the spoiler

flow: List[latch.types.metadata.FlowBase]#

Flow displayed in the spoiler card

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

Bases: object

Definition of a Fork branch

display_name: str#

String displayed in the fork’s multibutton

flow: List[latch.types.metadata.FlowBase]#

Child flow displayed in the fork card when the branch is active

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

Bases: latch.types.metadata.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, latch.types.metadata.ForkBranch]#

Mapping between branch keys to branch definitions. Order determines the order of options in the multibutton

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

Bases: object

Class for organizing parameter metadata

display_name: Optional[str] = None#

The name used to display the parameter on Latch Console

description: Optional[str] = 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: Optional[str] = None#

Whether this parameter should start a new section

placeholder: Optional[str] = None#

What should be rendered as a placeholder in the input box of the parameter before any value is inputed.

comment: Optional[str] = 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

appearance_type: latch.types.metadata.LatchAppearanceType = '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[latch.types.metadata.LatchRule]#

A list of LatchRule objects that inputs to this parameter must follow

detail: Optional[str] = None#
samplesheet: Optional[bool] = None#

Use samplesheet input UI. Allows importing from Latch Registry. Parameter type must be a list of dataclasses

property dict#
class latch.types.metadata.LatchMetadata(display_name: str, author: latch.types.metadata.LatchAuthor, documentation: typing.Optional[str] = None, repository: typing.Optional[str] = None, license: str = 'MIT', parameters: typing.Dict[str, latch.types.metadata.LatchParameter] = <factory>, wiki_url: typing.Optional[str] = None, video_tutorial: typing.Optional[str] = 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:

```python 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 name of the workflow

author: latch.types.metadata.LatchAuthor#

A LatchAuthor object that describes the author of the workflow

documentation: Optional[str] = None#

A link to documentation for the workflow itself

repository: Optional[str] = None#

A link to the repository where the code for the workflow is hosted

license: str = 'MIT'#

A SPDX identifier

parameters: Dict[str, latch.types.metadata.LatchParameter]#

A dictionary mapping parameter names (strings) to LatchParameter objects

wiki_url: Optional[str] = None#
video_tutorial: Optional[str] = None#
tags: List[str]#
flow: List[latch.types.metadata.FlowBase]#
no_standard_bulk_execution: bool = False#

Disable the standard CSV-based bulk execution. Intended for workflows that support an aleternative way of processing bulk data e.g. using a samplesheet parameter

property dict#

latch.types.utils module#

Module contents#