How can I dynamically set a flow owner and use it in a state handler to send a Slack alert on failure to that flow owner?

View in #prefect-server on Slack

Lana_Dann @Lana_Dann: is it possible to set a context variable on the flow level? for context, i want to set a flow owner to the flow and then use that value in my slack state handler. but even when i set

with prefect.context(dict(flow_owner="test")):

before defining the flow, i still get an error

Traceback (most recent call last):
  File "/Users/lanadann/.pyenv/versions/data-prefect-3.9.7/lib/python3.9/site-packages/prefect/engine/", line 161, in handle_state_change
    new_state = self.call_runner_target_handlers(old_state, new_state)
  File "/Users/lanadann/.pyenv/versions/data-prefect-3.9.7/lib/python3.9/site-packages/prefect/engine/", line 113, in call_runner_target_handlers
    new_state = handler(self.task, old_state, new_state) or new_state
  File "/Users/lanadann/prefect/data_prefect/lib/", line 13, in post_to_slack_on_failure
    f"@{prefect.context.flow_owner} "
AttributeError: 'Context' object has no attribute 'flow_owner'

Anna_Geller @Anna_Geller: perhaps you can set it in the README in the UI?

Lana_Dann @Lana_Dann: i don’t necessarily need the variable in the context, but i need a way to pass flow_owner into the slack state handler so that the automated slack messages can tag the owner in the alerts channel. can i do that through the readme?

Anna_Geller @Anna_Geller: are you on Prefect Cloud? is the flow owner also the person who registered the flow?
You could get the info about who created the flow automatically via a GraphQL query - it’s described here

Prefect Community: How can I identify changes made to a specific flow and who made those changes?

Lana_Dann @Lana_Dann: i am on prefect cloud! but we register flows in a CI script that runs in a repo so there’s nothing tying flows back to an individual

Kevin_Kho @Kevin_Kho: You can’t add stuff to the context like this for Cloud/Server runs as a “default” context. You can only add context for runs you trigger through the UI

Anna_Geller @Anna_Geller: agree with Kevin, I think your best bet is setting this info about flow owner as environment variable on your run config and retrieving this in your state handler this way

Kevin_Kho @Kevin_Kho: But I dont think this is meant for stuff like that. It’s more like you can change the value of today to backfill

Lana_Dann @Lana_Dann: hmm i assume you can’t add extra kwargs in the state handler either, right?
i think i can use the env variables set in the run config then… thanks for the help!

Anna_Geller @Anna_Geller: actually, I have a better idea! why not set your flow owner as a Parameter task? This would allow you to easily override it at runtime when needed and you could retrieve it in your state handler this way:



with Flow("yourflow") as flow:
    owner = Parameter("flow_owner", default="Lana")

Lana_Dann @Lana_Dann: that’s a great idea! thank you :smile:

Anna_Geller @Anna_Geller: Lana, one caveat though: if you don’t use this owner object in downstream tasks, you will need to manually add it to your flow like so:


(just to save you some trouble)

Lana_Dann @Lana_Dann: thank you again! that’s actually exactly what i did just now. we have a BaseECSFlow method that returns a flow with all of the metadata set up to run in our default ECS cluster. i updated the logic to:

    ecs_flow = Flow(
    ecs_flow.add_task(Parameter("flow_owner", default=owner))

    return ecs_flow

and now folks can use the flow like this:

with BaseECSFlow(
) as flow:

Kevin_Kho @Kevin_Kho: You can also call it

with Flow(..) as flow:
    flow_owner = Parameter("flow_owner", "default")()

notice the second (). This will make it used I think