Adding a SQLite Trigger to a Lattice#

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

Prerequisites#

Import Covalent and the trigger.

[10]:

import covalent as ct
from covalent.triggers import SQLiteTrigger

Procedure#

  1. Create a SQLiteTrigger object that performs a trigger.

[12]:

sqlite_trigger = SQLiteTrigger(db_path='path/to/your/database.sqlite',table_name='table name')

  1. Create a workflow:

[13]:

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

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

[14]:

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

5041f1fc-8943-4b96-9f26-a3c7f35cabef

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

[15]:

sqlite_trigger.lattice_dispatch_id = dispatch_id
sqlite_trigger.register()

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

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

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

[16]:

ct.stop_triggers(dispatch_id)

[2023-09-22 07:37:27,218] [DEBUG] local.py: Line 334 in stop_triggers: Triggers for following dispatch_ids have stopped observing:
[2023-09-22 07:37:27,220] [DEBUG] local.py: Line 336 in stop_triggers: 5041f1fc-8943-4b96-9f26-a3c7f35cabef

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 Time Trigger to a Lattice

Adding a Database Trigger to a Lattice