How to add callable function dependencies to an electron

In this section, we show how to add functions that are invoked before or after executing an electron. First, we import covalent.

import covalent as ct
from pathlib import Path

Next, we define the functions to be executed before and after the electron:

def call_before_hook(filename):
    Path(filename).write_text('Hello world!')
    return filename

def call_after_hook(filename):
    Path(filename).unlink()

We then declare the call_before and call_after variables in the electron to reference the corresponding functions which are passed as arguments to DepsCall():

@ct.electron(
    call_before=ct.DepsCall(call_before_hook, args=('test.txt',), retval_keyword='my_file'),
    call_after=ct.DepsCall(call_after_hook, args=('test.txt',)),
)
def read_from_file(my_file=None):
    with open(my_file,'r') as f:
        return f.read()

@ct.lattice
def workflow():
    return read_from_file()

dispatch_id = ct.dispatch(workflow)()
r = ct.get_result(dispatch_id, wait=True)
print(r.result)

Pne can pass other types of Deps as variables to call_before and call_after:

from covalent import DepsBash, DepsPip

deps_pip=DepsPip(packages=["numpy==1.22.4"])
deps_bash=DepsBash(commands=["echo $PATH >> /tmp/path.txt"])

@ct.electron(
    call_before=[deps_pip],
    call_after=[deps_bash],
)
def my_task():
    pass