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)