Available in latch >= 2.62.0
- Call the workflow with Python parameter values.
- Use a
LaunchPlan from a previously registered workflow.
Pre-requisites
- Please make sure you have registered your workflow to Latch using
latch register. This creates a new workflow version, which you will reference in your script.
Usage
from latch.types.file import LatchFile
from latch_cli.services.launch.launch_v2 import launch
from wf.entrypoint import GenomeReference, Sample
# Using python parameter values
execution_id = launch(
wf_name="my_workflow",
version="0.1.0",
params={
"input": [
Sample(
patient="TestSample",
fastq_1=LatchFile(
"latch:///test_L1_1.fastq.gz"
),
...
),
...
],
"genome": GenomeReference.GATK_GRCh37,
...
}
)
wf_name: Name of the workflow function. This is the function defined immediately after the @workflow decorator in your code (Example)version: Workflow version registered on Latchparams: Dictionary of parameter names and values. If the python values are not typed using the exact same types used in the workflow function signature (including the module paths), the best_effort parameter should be set to True.best_effort (default True): (available in latch >= 2.67.8) When set to True, allows for flexible conversion of params to workflow inputs. This enables launching workflows outside of the workflow environment with compatible values (e.g., strings for enums where the string matches an enum option, or dictionaries/dataclasses with all the fields of a particular dataclass).
import email.utils
import time
from datetime import datetime
from latch_cli.services.launch.launch_v2 import launch_from_launch_plan
from latch_cli.tinyrequests import post
from latch_cli.utils import get_auth_header
from latch_sdk_config.latch import config
# Using a previously registered LaunchPlan
execution_id = launch_from_launch_plan(
wf_name="my_workflow",
version="0.1.0",
lp_name="Test Data",
)
# Polling execution status
while True:
list_resp = post(
url=config.api.execution.list,
headers = {"Authorization": get_auth_header()},
json={"ws_account_id": "XXXXX"},
).json()
target_execution = list_resp[str(execution_id)]
target_execution_status: str = target_execution.get('status')
if target_execution_status == 'SUCCEEDED':
break
elif target_execution_status == 'FAILED':
raise Exception("Execution failed")
elif target_execution_status == 'ABORTED':
raise Exception("Execution aborted")
elif target_execution_status == 'UNDEFINED':
time.sleep(5)
continue
start_time_str: str = target_execution.get('start_time')
start_time_tuple = email.utils.parsedate_tz(start_time_str)
if start_time_tuple is None:
raise Exception("Failed to parse start time")
start_time = datetime.fromtimestamp(email.utils.mktime_tz(start_time_tuple))
current_time = datetime.now()
if (current_time - start_time).total_seconds() > 300:
raise TimeoutError("Execution has been running for more than 5 minutes")
time.sleep(10)
wf_name: Name of the workflow function. This is the function defined immediately after the @workflow decorator in your code (Example)version: Workflow version registered on Latchlp_name: Name of the LaunchPlan (Example)
Important note on Python version: If the Python version in the workflow’s
Docker image does not match the Python version running the script that calls
launch or launch_from_launch_plan, the function may fail with a dill
unpickling error. In this case, it is recommended to use the same Python
version used during workflow registration.
