Hey all, I'm attempting to use a state handler to ...
# ask-community
j
Hey all, I'm attempting to use a state handler to update a database for Flow state transitions to Running, Failed, and Successful. I need to use a Flow parameter result to make the update, which I can retrieve from the state result for Failed and Successful, but its not clear to me how to do this for the Running state, which has NoResultType
Copy code
def handle_state(flow, old_state, new_state):
    def param_value_for(param_name, state):
        p = next(filter(lambda k: k.name == param_name, new_state.result.keys()))
        r = state.result[p]
        return r.result

    if new_state.is_running():
        foo = param_value_for('foo', new_state)
        print('Starting to run')
    elif new_state.is_failed():
        foo = param_value_for('foo', new_state)
        print(f'We have a failure for {foo}')
    elif new_state.is_successful():
        foo = param_value_for('foo', new_state)
        print(f'We have a success for {foo}')
c
Hey @Jason Damiani - take a look in
prefect.context
for the flow parameters:
Copy code
p = prefect.context.get("parameters", {}).get(param_name)
this should be populated at the time of your flow’s state handlers being called
j
ah cool, thanks. I figured there must be an easier way
c
yea np!
j
Are parameters not documented here because they aren't always available in the context? https://docs.prefect.io/api/0.6.0/utilities/context.html#context-2
c
That’s a great question; they should always be in context during a flow run, so that’s an oversight! I’ll update
j
@Chris White maybe we should also start populating that list with all the Cloud variables that are available as well? Run ID, etc.
c
yea that’s a good idea, i’ll get that going this afternoon
j
👍
c
@Marvin archive “Issue accessing Flow Parameters from within a state handler”