How to add a directory trigger to a lattice

In this guide we’ll illustrate how to use a DirTrigger to trigger workflow runs automatically whenever a file in a directory gets modified.

Let’s first import the required parts:

import covalent as ct
from covalent.triggers import DirTrigger
from pathlib import Path

For our test case, let’s create a new file in our current directory named my_text_file.txt and have 10 numbers in it.

with open("my_text_file.txt", "w") as f:
    for i in range(10):
        f.write(f"{i}\n")

Now, let’s create our DirTrigger object which performs a trigger action whenever something in the current directory gets modified.

dir_trigger = DirTrigger(dir_path=str(Path(".").resolve()), event_names="modified")

Let’s create a simple workflow now and assign this trigger to the lattice:

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

Once we’ve made sure that the covalent server is running, we can perform the dispatch for my_workflow as such:

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

fb6235ea-248c-46b7-a7b7-76a7f7ad5da3

Now, if you check the UI you’ll see that a new my_workflow gets dispatched whenever we add or remove a number from my_text_file.txt. If we want to stop that after a while, we can use the ct.stop_triggers function:

ct.stop_triggers(dispatch_id)

Triggers for following dispatch_ids have stopped observing:
fb6235ea-248c-46b7-a7b7-76a7f7ad5da3

The above will prevent any new dispatches from happening due to the trigger action on my_workflow lattice.