https://prefect.io logo
Channels
Powered by
p

Philip MacMenamin

11/17/2020, 9:23 PM
How do I ask a task what its terminal state is?
j

Jenny

11/17/2020, 9:37 PM
Hi @Philip MacMenamin - can you give a bit more information about what you mean here? When you say "ask a task" are you meaning an API query? Or something else?
p

Philip MacMenamin

11/17/2020, 9:39 PM
hi Jenny - yes,
Copy code
@task(name="a")
def add_one(x):
    return x + 1

with Flow("Add One") as flow:
    a = add_one(1)
I'd like to be able to ask what Task a's final state is, eg if it's successful, or not.
j

Jenny

11/17/2020, 9:52 PM
Not sure how you want to ask but a state handler with .is_finished() might help you: https://docs.prefect.io/core/concepts/notifications.html#sending-a-simple-notification The link here is for notifications but you can use state-handlers in other ways too.
p

Philip MacMenamin

11/17/2020, 10:02 PM
so you need to add a state handler to see if your tasks have completed successfully? There isn't some default way to ask if a normal task completed successfully?
j

Jenny

11/17/2020, 10:08 PM
I'm still not clear what it is you're asking here - Can you give me a bit more info about what it is you want to do? Do you want to see the task run state in your logs? In the UI? You should be able to see those without having to add anything. Example logs from a simple flow (run locally using flow.run() - you can see the final task run states:
Copy code
[2020-11-17 17:06:25-0500] INFO - prefect.FlowRunner | Beginning Flow run for 'simple-map'
[2020-11-17 17:06:25-0500] INFO - prefect.TaskRunner | Task 'c': Starting task run...
[2020-11-17 17:06:25-0500] INFO - prefect.TaskRunner | Task 'c': Finished task run for task with final state: 'Success'
[2020-11-17 17:06:25-0500] INFO - prefect.TaskRunner | Task 'return_list': Starting task run...
[2020-11-17 17:06:25-0500] INFO - prefect.TaskRunner | Task 'return_list': Finished task run for task with final state: 'Success'
[2020-11-17 17:06:25-0500] INFO - prefect.TaskRunner | Task 'mapped_task': Starting task run...
[2020-11-17 17:06:25-0500] INFO - prefect.TaskRunner | Task 'mapped_task': Finished task run for task with final state: 'Mapped'
[2020-11-17 17:06:25-0500] INFO - prefect.TaskRunner | Task 'mapped_task[0]': Starting task run...
[2020-11-17 17:06:25-0500] INFO - prefect.mapped_task[0] | 1
[2020-11-17 17:06:25-0500] INFO - prefect.TaskRunner | Task 'mapped_task[0]': Finished task run for task with final state: 'Success'
[2020-11-17 17:06:25-0500] INFO - prefect.TaskRunner | Task 'mapped_task[1]': Starting task run...
[2020-11-17 17:06:25-0500] INFO - prefect.mapped_task[1] | 2
[2020-11-17 17:06:25-0500] INFO - prefect.TaskRunner | Task 'mapped_task[1]': Finished task run for task with final state: 'Success'
[2020-11-17 17:06:25-0500] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded
Or if you want to use the API you can query for that task's task runs (or flow run and get its task runs).
Copy code
query {
  task_run (where: {task: {name: {_eq: "a"}}}) {
    state
  }
}

query {
  flow_run (where: {flow: {name: {_eq: "Add One"}}}) {
    task_runs (where: {task: {name: {_eq: "a"}}}) {
      state
      name
  }
  }
}
p

Philip MacMenamin

11/17/2020, 10:24 PM
I'd like something like this:
Copy code
with Flow("example") as flow:
    a = add_one(1)
    b = task_2(2, task_args=dict(name="b"))
    if a.is_successful and b.is_successful:
        do_something(b)
as in, at the end of every run, I'd like to just go through a set of tasks, ask for their final status and take some action based on the final task states.
Alternatively, if the flow is considered to have been successful (eg the above case where all ref tasks succeeded), then I'd like to get that 
flow_run_id
and take some action. I want a way to notify some external entity that a run has completed, and it's succeeded or failed.
j

Jenny

11/17/2020, 10:51 PM
If you want to notify an external entity based on state then a state handler as a notifier is actually going to work well for you. (You can add to a flow or a task). If you want to take certain actions depending on upstream task results, you could also consider triggers: https://docs.prefect.io/api/latest/triggers.html#triggers If you just want to examine final states, then the docs here should help: https://docs.prefect.io/core/getting_started/first-steps.html
p

Philip MacMenamin

11/17/2020, 11:39 PM
ok - thanks, I'll have a look
Powered by