https://prefect.io logo
n

Noam polak

02/14/2022, 8:07 AM
Hey everyone I have a flows that triggers other child flows and I want to notify slack when one of them fails- this is my ( logic) code:
Copy code
child_flow = create_flow_run(
            flow_name=CHILD_FLOW,
            parameters={
                "input_data": input_data,
                "run_id": run_id,
            },
            project_name="default",
        )
child_result = get_task_run_result(
            child_flow,
            task_slug="child_flow-copy",
            poll_time=3,
        )
I tried to add handler to get_task_run_result but it get error:
Copy code
child_result = get_task_run_result(
            hazarder_flow,
            task_slug="child_flow-copy",
            poll_time=3,
            state_handlers=[post_to_slack_task_handler],
        )

TypeError: got an unexpected keyword argument 'state_handler'
So how can I do it? thanks
k

Kevin Kho

02/14/2022, 2:38 PM
Hi @Noam polak,the state handler should be defined in the subflow. You can’t define it like this at runtime. Or you can try:
Copy code
from prefect... import get_task_run_result

get_task_run_result.state_handlers=[post_to_slack_task_handler]
to change the state handler of that import
n

Noam polak

02/14/2022, 3:19 PM
Hey, Thanks I'll try your second option
k

Kevin Kho

02/14/2022, 3:23 PM
Actually, what you want is:
Copy code
with Flow(..) as flow:
    create_flow_run(...)
    wait_for_flow_run(..., raise_final_state=True)
    get_task_run_result(...)
and then you want to modify the state handler for
wait_for_flow_run
You need it anyway to guarantee
create_flow_run
will finish before you try to get the result
n

Noam polak

02/16/2022, 8:07 AM
From what I read get_task_run_result always waits for child flow to finish, isn't it?
k

Kevin Kho

02/16/2022, 2:12 PM
Ah yeah you are right. It does poll to wait for completion
n

Noam polak

02/16/2022, 3:15 PM
Thanks, your answer helped get what I needed 🎉
3 Views