c

    Christian

    1 year ago
    Hi. 👋 I want to access prefect.context.parameters from within a task state_handler but the value I get is always
    None
    . 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? Thanks
    j

    josh

    1 year ago
    @Christian could you post an example of your state handler? And the parameters being passed into your flow
    c

    Christian

    1 year ago
    Sure (by the way, prefect.context.flow_name has a valid entry)
    def 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]
    )
    Parameter (defined in the scope of the flow, before the task is called):
    notification_email = Parameter(      "notification_email", default="<mailto:an_email@gmail.com|an_email@gmail.com>")
    j

    josh

    1 year ago
    Just wondering, does accessing it from
    prefect.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! 🙂
    c

    Christian

    1 year ago
    Thanks. I'll have a look at triggers.