View in #prefect-server on Slack
@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/runner.py", 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/task_runner.py", 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/notifiers.py", line 13, in post_to_slack_on_failure
f"@{prefect.context.flow_owner} "
AttributeError: 'Context' object has no attribute 'flow_owner'
@Anna_Geller: perhaps you can set it in the README in the UI?
@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: 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: 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: 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: 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: 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: 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: 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:
prefect.context["parameters"].get("flow_owner")
assuming:
with Flow("yourflow") as flow:
owner = Parameter("flow_owner", default="Lana")
@Lana_Dann: that’s a great idea! thank you 
@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:
flow.add_task(owner)
(just to save you some trouble)
@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(
name=name,
schedule=schedule,
run_config=run_config,
storage=get_default_storage(file_path),
)
ecs_flow.add_task(Parameter("flow_owner", default=owner))
return ecs_flow
and now folks can use the flow like this:
with BaseECSFlow(
name=FLOW_NAME,
file_path=FILE_PATH,
start_date=START_DATE,
interval=INTERVAL,
owner=OWNER
) as flow:
...
@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