Lana Dann
04/05/2022, 11:02 PMwith 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
04/05/2022, 11:08 PMLana Dann
04/05/2022, 11:10 PMflow_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
04/05/2022, 11:11 PMLana Dann
04/05/2022, 11:13 PMKevin Kho
04/05/2022, 11:14 PMAnna Geller
04/05/2022, 11:15 PMKevin Kho
04/05/2022, 11:15 PMtoday
to backfillLana Dann
04/05/2022, 11:15 PMAnna Geller
04/05/2022, 11:21 PMprefect.context["parameters"].get("flow_owner")
with Flow("yourflow") as flow:
owner = Parameter("flow_owner", default="Lana")
Lana Dann
04/05/2022, 11:22 PMAnna Geller
04/05/2022, 11:27 PMflow.add_task(owner)
(just to save you some trouble)Lana Dann
04/05/2022, 11:28 PMBaseECSFlow
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
04/05/2022, 11:29 PMwith Flow(..) as flow:
flow_owner = Parameter("flow_owner", "default")()
notice the second ()
. This will make it used I thinkLana Dann
04/06/2022, 5:19 PMKevin Kho
04/06/2022, 5:25 PM<https://cloud.prefect.io/[tenant_name]/flow-run/[flow_run_id]>
The flow_run_id is easy to grab but tenant name/slug is not.
Try using the Client().get_defrault_tenant_slug
and then constructing it?Lana Dann
04/06/2022, 5:30 PMKevin Kho
04/06/2022, 5:32 PMflow_run_id
from contextLana Dann
04/06/2022, 7:37 PMKevin Kho
04/06/2022, 7:41 PMwith Flow(..., state_handlers=[..]) as flow:
will work and the state handler can be of signature
def myhandler(obj: Union[Flow, Task], old_state, new_state)
meaning the same handler can be used at the task level and flow levelLana Dann
04/06/2022, 7:43 PMprefect.context.task_name
if i use the state handler in the flow?Kevin Kho
04/06/2022, 7:45 PMLana Dann
04/06/2022, 7:47 PMKevin Kho
04/06/2022, 8:00 PMtask_run_count
in the state handler and an if else to not repeatedly firetask_run_count
to max_retries
(which is not in context) before firing the Slack alerttask.max_retries
inside the state handler if the signature is
def myhandler(state, old_state, new_state):
and compare it to
task_run_count