Adding a Time Trigger to a Lattice

This example illustrates how to use a covalent.trigger.TimeTrigger to trigger workflow dispatches automatically at a specified interval.

Prerequisites

Import Covalent and the trigger.

import covalent as ct
from covalent.triggers import TimeTrigger

Procedure

1. Create a TimeTrigger object that performs a trigger action every 5 seconds.

time_trigger = TimeTrigger(time_gap=5)

2. Create a workflow:

@ct.lattice
@ct.electron
def my_workflow():
    return 42

3. Dispatch my_workflow, disabling its first execution using the disable_run parameter in ct.dispatch.

dispatch_id = ct.dispatch(my_workflow)()
print(dispatch_id)

44f56fcf-96dc-4089-84d4-069fd13e3e58

4. Attach the trigger to the dispatch_id and register it with the trigger server.

time_trigger.lattice_dispatch_id = dispatch_id
time_trigger.register()

5. Monitor the Covalent UI. Watch the Dashboard for new dispatches of my_workflow.

6. In the Covalent UI, observe that a new my_workflow is dispatched every five seconds.

7. To disable triggers on the dispatch, use the ct.stop_triggers function.

ct.stop_triggers(dispatch_id)

[2023-09-25 08:51:25,893] [DEBUG] local.py: Line 334 in stop_triggers: Triggers for following dispatch_ids have stopped observing:
[2023-09-25 08:51:25,894] [DEBUG] local.py: Line 336 in stop_triggers: 44f56fcf-96dc-4089-84d4-069fd13e3e58

Note that the stop_triggers function disables all triggers attached to the specified dispatch.

See Also

Adding a Directory Trigger to a Lattice

Adding a SQLite Trigger to a Lattice

Adding a Database Trigger to a Lattice