Skip to main content
This tutorial walks through a practical example of integrating Latch Registry into a workflow.

​
What You Will Achieve

By the end of this tutorial, your workflow will let scientists seamlessly connect Registry tables to workflow execution. Here’s the user experience you are enabling:
  1. In the UI, scientists click “Import from Registry”, choose a table, and select specific rows.
  2. A modal appears to match Registry column names with the workflow input parameters you define in code.
  3. Once confirmed, the selected rows are imported directly into the workflow.
  4. Users can launch the workflow with the imported data.
  5. The workflow then upserts results back into the same Registry records.
  6. Updated records are visible in the original Registry table, linking inputs with outputs.

​
Importing rows from Registry

​
Usage Example

from dataclasses import dataclass

from latch.resources.tasks import small_task
from latch.resources.workflow import workflow
from latch.types.file import LatchFile
from latch.types.metadata import LatchAuthor, LatchMetadata, LatchParameter
from latch.types.samplesheet_item import SamplesheetItem

# Define a dataclass to represent a row in the samplesheet
@dataclass
class Row:
    sample_name: str
    fastq_1: LatchFile
    fastq_2: LatchFile


metadata = LatchMetadata(
    display_name="Test Record from Samplesheet Input",
    author=LatchAuthor(
        name="CHANGE ME",
    ),
    parameters={
        "rows": LatchParameter(
            display_name="Test",
            samplesheet=True, # Enable samplesheet input display in the UI
            batch_table_column=True,  # Show this parameter in batched mode.
        ),
    },
)

@small_task
def task(rows: list[SamplesheetItem[Row]]) -> None: # Wrap the `Row` dataclass in `SamplesheetItem` 
    for row in rows:
        if row.record is None: # Check if the row was imported from Registry
            print(f"{row.data.sample_name}: Data is not from registry")
            continue

        # If the row was imported from Registry, we can access the Record object

        print(f"{row.data.sample_name}:")
        print(f" -> From Record {row.record.id} in Table {row.record.get_table_id()}")
        print(f" -> Created at {row.record.get_creation_time().isoformat()}")
        print(f" -> Last updated at {row.record.get_last_updated().isoformat()}")


@workflow(metadata)
def fancy_samplesheet_wf(rows: list[SamplesheetItem[Row]]) -> None:
    return task(rows=rows)

​
Core Ideas

To import a table from Registry into your workflow, follow these steps:
  1. Define a dataclass: Create a Python dataclass that represents a single row in the samplesheet.
  2. Mark it as a samplesheet: In the LatchMetadata, set samplesheet=True to indicate that this dataclass should be treated as a samplesheet.
  3. Match Registry columns: The fields in your dataclass should match the column names in the Registry table (or be a subset). If your dataclass fields use the exact same names as the Registry columns, the workflow UI will automatically align them in the “column matching” modal.
  4. Wrap rows in SamplesheetItem: Each row should be wrapped in a SamplesheetItem, which allows the workflow to detect whether the row came from Registry.
  5. A SamplesheetItem has two fields:
    • data – the actual row data. For example, row.data.sample_name will give you the value from the sample_name column.
    • record – the Registry Record object for that row.
      • If the row was imported from Registry, record will be populated with the corresponding Record.
      • If the row was created manually (not imported), record will be None.

​
Upserting rows to Registry

A Table can be modified by using the Table.update() function. Table.update() returns a context manager (and hence must be called using with syntax) with an upsert_record method. Visit the Table Object page for the API reference.

​
Usage Example

from latch.registry.table import Table
from latch.types.directory import LatchDir

@small_task
def task(rows: list[SamplesheetItem[Row]]) -> None: # Wrap the `Row` dataclass in `SamplesheetItem` 
    for row in rows:
        if row.record is None: # Check if the row was imported from Registry
            print(f"{row.data.sample_name}: Data is not from registry")
            continue

        # If the row was imported from Registry, we can access the Record object

        print(f"{row.data.sample_name}:")
        print(f" -> From Record {row.record.id} in Table {row.record.get_table_id()}")
        print(f" -> Created at {row.record.get_creation_time().isoformat()}")
        print(f" -> Last updated at {row.record.get_last_updated().isoformat()}")

        # Imagine a directory object that we want to upsert
        cellranger_output = LatchDir(...)

        # Upsert the record
        table = Table(id=row.record.get_table_id())
        with table.update() as updater:
            updater.upsert_record(
                "<Record identifier under the Name column in Registry table UI>",
                cellranger_output=cellranger_output
            )

​
Core Ideas

  • Retrieve the Table ID by calling get_table_id() on the Record object.
  • Initialize a Table object by calling Table(id=table_id).
  • Call Table.update() to get a context manager with an upsert_record method.
⌘I