@task
def get_values():
    return ["value", "test", "demo"]

@task(task_run_name="{val}", state_handlers=[post_to_mail])
def compute(val):
    if val == "demo":
        raise ValueError("Nope!")

with Flow("task_run_names") as flow:
    vals = get_values()
    compute.map(vals)

from prefect import task, Flow
from prefect import Flow, task
from prefect.engine.state import Failed
from prefect.client.secrets import Secret
import smtplib, ssl
import prefect

port = 587  # For starttls
smtp_server = "<http://smtp.gmail.com|smtp.gmail.com>"
sender_email=Secret('EMAIL_USERNAME')
sender_email=sender_email.get()

receiver_email=Secret('receiver_email')
receiver_email=receiver_email.get()

password=Secret('EMAIL_PASSWORD')
password=password.get()

def post_to_mail(task, old_state, new_state):
    if new_state.is_failed():
        flow_run_id = prefect.context.get("flow_run_id")
        flow_id = prefect.context.get("flow_id")
        msg = "Task {0} failed and is retrying at {1}""".format(task, new_state)
        message = "Subject: {0}\n\n{1}".format(flow_id, msg)
        context = ssl.create_default_context()
        with smtplib.SMTP(smtp_server, port) as server:
            server.starttls(context=context)
            server.login(sender_email, password)
            for i in receiver_email:
                server.sendmail(sender_email, i, message)

    if new_state.is_successful():
        flow_run_id = prefect.context.get("flow_run_id")
        flow_id = prefect.context.get("flow_id")
        msg = "Task {0} failed and is retrying at {1}""".format(task, new_state)
        message = "Subject: {0}\n\n{1}".format(flow_id, msg)
        context = ssl.create_default_context()
        with smtplib.SMTP(smtp_server, port) as server:
            server.starttls(context=context)
            server.login(sender_email, password)
            for i in receiver_email:
                server.sendmail(sender_email, i, message)

task_run_id

task_run_name

query = """
query {task_run_by_pk (id: "5144f2c0-a29c-4182-a06f-4402341aebda") {
  name
}}
"""
client.graphql(query)