shijas km
02/02/2022, 2:50 PMKevin Kho
shell = ShellTask(...)
with Flow(...) as flow:
s = shell()
something_else(upstream_tasks=[s])
if you really need it before the flow begins, you need to use it in a State Handler . This is not a task though. It’s just like a Python function.
def mystatehandler(obj,old_state,new_state):
if new_state.is_running():
ShellTask().run(...)
with Flow(..., state_handlers=[mystatehandler]) as flow:
....
shijas km
02/02/2022, 2:59 PMNate
02/02/2022, 2:59 PMShellTask
and normal prefect tasks
from prefect import Flow, task
from prefect.tasks.shell import ShellTask
@task(log_stdout=True)
def regular_task():
print('hi')
my_shell_task = ShellTask(
helper_script="cd ~",
)
with Flow("My Flow") as flow:
# task will be executed in home directory
contents = my_shell_task(command='ls')
regular_task()
if __name__ == "__main__":
flow.run()
shijas km
02/02/2022, 2:59 PMKevin Kho
shijas km
02/02/2022, 3:23 PMKevin Kho
shijas km
02/02/2022, 3:32 PMKevin Kho