latch.functions package

Submodules

latch.functions.messages module

latch.functions.messages.message(typ: str, data: Dict[str, Any]) → None

Display a message prominently on the Latch console during and after a task execution.

The Latch platform first processes this message internally, then displays it under your task’s execution page.

Parameters:

• typ – A message type that determines how your message is displayed. Currently one of ‘info’, ‘warning’, or ‘error’.

• data – The data displayed on the Latch console, formatted as follows: `{'title': ..., 'body': ...}`.

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

    ...

latch.functions.operators module

DEPRECATED

Mimics channel operators from Nextflow, using the correspondence Channel –> Python Dictionary

latch.functions.operators.left_join(left: Dict[str, Any], right: Dict[str, Any]) → Dict[str, Any]

A standard left join of two dictionaries, joining on their keys

latch.functions.operators.right_join(left: Dict[str, Any], right: Dict[str, Any]) → Dict[str, Any]

A standard right join of two dictionaries, joining on their keys

latch.functions.operators.inner_join(left: Dict[str, Any], right: Dict[str, Any]) → Dict[str, Any]

A standard inner join of two dictionaries, joining on their keys

latch.functions.operators.outer_join(left: Dict[str, Any], right: Dict[str, Any]) → Dict[str, Any]

A standard outer join of two dictionaries, joining on their keys

latch.functions.operators.group_tuple(channel: List[Tuple], key_index: int | None = None) → List[Tuple]

Operator to mimic the groupTuple construct from Nextflow:

The groupTuple operator collects tuples (or lists) of values emitted by the source channel grouping together the elements that share the same key. Finally it emits a new tuple object for each distinct key collected.

Parameters:

• channel – A list of tuples to be grouped by key_index

• key_index – Which index of the tuple to match against - if not provided, defaults to 0

Example

>>> 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)
[(1, ['A', 'B', 'C']), (2, ['C', 'A']), (3, 'B', 'D')]

latch.functions.operators.latch_filter(channel: List[Any], predicate: Callable | Pattern | type | None) → List[Any]

Filters a given list with either a predicate, a regex, or a type

latch.functions.operators.combine(channel_0: List[Any], channel_1: List[Any], by: int | None = None) → List | Dict[str, List[Any]]

Creates a Cartesian product of the two provided channels, with the option to first group the elements of the channels by a certain index and then doing individual products on the groups.

Parameters:

• channel_0 – A list. If by is provided, all elements must be tuples of the same length

• channel_1 – A list. If by is provided, all elements must be tuples of the same length as channel_0

• by – If provided, which index to group by first.

Example

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

latch.functions.secrets module

latch.functions.secrets.get_secret(secret_name: str)

A utility to allow users to reference secrets stored in their workspace on Latch.

Important: When running an execution locally, whether on your own computer or using latch develop, the only secrets you will be able to access are the ones in your personal workspace. To use secrets from a shared workspace, register your workflow and run it on Latch.

Examples

>>> get_secret("test-secret")
"test-value-123"

Module contents