https://prefect.io logo
m

Milly gupta

02/08/2021, 11:31 PM
Hi All, does the flow object has task state information stored as well. I want to use state handler at flow level but would like to send task level msg if there were any task failures in the flow.
k

Kyle Moon-Wright

02/08/2021, 11:54 PM
Hey @Milly gupta, We could add State Handlers at both the Task and Flow Level: where the Task level State Handler adds something to Prefect Context (the underlying metadata about your Flow Run) upon failure and the Flow Level State Handler retrieves that information from Context to send in an email upon a Flow’s failure. More information on Prefect Context and adding stuff to it here: https://docs.prefect.io/core/concepts/execution.html#context Hopefully others have encountered this scenario and have insight! edit: I’ve learned manually adding data to Prefect Context is considered an anti-pattern and will generally not work
m

Milly gupta

02/08/2021, 11:58 PM
Thanks Kyle. Yes I have tried that by adding in context at task level but when I access that prefect context at flow level state handler for the key I set at task level, I get None.
k

Kyle Moon-Wright

02/09/2021, 12:07 AM
Hmm yes, apologies - actually adding to Prefect Context manually may be considered an anti-pattern. In this scenario, we may need to query the API (assuming you’re using a backend API rather than
flow.run()
) to get all task state information for the Flow Run. Otherwise, either a task level or flow level state handler will have to suffice, and not this Context stuff as previously suggested.
m

Milly gupta

02/09/2021, 12:10 AM
I am just running via Prefect Cloud UI atm.
Which API can I use to get all task state?
k

Kyle Moon-Wright

02/09/2021, 12:17 AM
Awesome! You can make a GraphQL query for this information in a variety of ways, but I’d recommend testing out your query in the Interactive API first to ensure you’re fetching the right information.
m

Milly gupta

02/09/2021, 12:20 AM
Can you provide an example please how can I get state of a task from a flow state handler
k

Kyle Moon-Wright

02/09/2021, 12:39 AM
Here is an example GQL query you can make, though you’ll need to adjust it for your purposes:
Copy code
query {
    flow_run(
        where: {
            state: {_eq: "Failed"},
            flow_id: {eq: "MY_FLOW_ID"},
        }
    ) {
        task_runs(
          where: {state: {_eq: "Failed"}}
        ){
            name,
      		state_message,
      		created
        }
    }
}
m

Milly gupta

02/09/2021, 1:35 AM
thanks Kyle.
👍 1
when I use above query with client.graphql , I am getting Exception raised while calling state handlers: ClientError([{'path': ['flow_run'], 'message': 'field "flow_run" not found in type: \'query_root\'', 'extensions': {'path': '$.selectionSet.flow_run', 'code': 'validation-failed', 'exception': {'message': 'field "flow_run" not found in type: \'query_root\''}}}])
Is it the formatting error?
k

Kyle Moon-Wright

02/09/2021, 1:41 AM
Hmm, not sure. First thing I would check is the USER token and that it is provisioned correctly, then I would check query formatting. For formatting, despite the queries in the documentation, I tend to write them with multi-line strings which makes things easier for me to read - as an example:
Copy code
client.graphql(
    """
    query{
        flow{
            name,
            ...
        }
    }
    """
)
m

Milly gupta

02/09/2021, 1:42 AM
Yeah that's what I did.
and how can I get a flow id?
k

Kyle Moon-Wright

02/09/2021, 1:50 AM
You can query for it or find it in the URL of the flow screen
m

Milly gupta

02/09/2021, 1:52 AM
query via API?
👍 1
Now I am getting Exception raised while calling state handlers: HTTPError('400 Client Error: Bad Request for url: https://api.prefect.io/graphql')
k

Kyle Moon-Wright

02/09/2021, 2:24 AM
For that one, we may need to make sure our query is correct but it’s tough to say for sure.
Apologies, I’m off for the night! 😴 I hope you can make some progress on this, and if not we can tackle it tomorrow!
3 Views