Christian
12/16/2020, 10:27 PMNone
.
The idea is to send an email (EmailTask) on a task failure and I want to parameterise the email address that will be alerted. Is there another way to pass a variable into a state_handler? Am I going in the wrong direction?
Thanksjosh
12/16/2020, 10:31 PMChristian
12/16/2020, 10:46 PMdef email_on_failure(task, old_state, new_state):
"""Send email on FAIL state"""
if new_state.is_failed():
notification_email = prefect.context.parameters.get("notification_email")
flow_name = prefect.context.flow_name
task = EmailTask(
subject="Prefect alert: {} {}".format(flow_name, new_state),
msg=html.escape(
"{} GreatExpectation Validation failed\n\nTask: {}; New State: {}".format(
flow_name, task, new_state
)
),
email_from="<mailto:server@gmail.com|server@gmail.com>",
email_to=notification_email,
smtp_server="some.server",
smtp_port=25,
smtp_type="STARTTLS",
).run()
and:
validation_task = RunGreatExpectationsValidation(
state_handlers=[email_on_failure]
)
notification_email = Parameter( "notification_email", default="<mailto:an_email@gmail.com|an_email@gmail.com>")
josh
12/16/2020, 10:59 PMprefect.context.get("parameters", {}).get(param_name)
have any effect? Going off of this issue here https://github.com/PrefectHQ/prefect/issues/1307. If it doesn’t then would you mind opening an issue in the repo with a reproducible example so we can triage?
Also it seems like you’re using an EmailTask inside that handler for failed states and instead I would recommend using the EmailTask in your flow directly as a downstream to the task you are checking to fail. And I would suggest taking advantage of triggers to accomplish this same behavior!
https://docs.prefect.io/core/getting_started/next-steps.html#triggers
https://docs.prefect.io/api/latest/triggers.html
That way you have the email task as part of your flow and you can react specifically to the failure of the upstream task! 🙂Christian
12/16/2020, 11:09 PM