Repository: apache/airflow
File: task-sdk/src/airflow/sdk/execution_time/task_runner.py (Python)

Code before:
    default_html_content = (
        "Try {{try_number}} out of {{max_tries + 1}}<br>"
        "Exception:<br>{{exception_html}}<br>"
        'Log: <a href="{{ti.log_url}}">Link</a><br>'
        "Host: {{ti.hostname}}<br>"
        'Mark success: <a href="{{ti.mark_success_url}}">Link</a><br>'
    )

    # Add exception_html to context for template rendering
    exception_html = str(error).replace("\n", "<br>")
    additional_context = {
        "exception": error,
        "exception_html": exception_html,
        "try_number": ti.try_number,
        "max_tries": ti.max_tries,
    }
    email_context = {**context, **additional_context}
    to_emails = task.email
    if not to_emails:
        return

    try:
        notifier = SmtpNotifier(
            to=to_emails,
            subject=default_subject,
            html_content=default_html_content,
            from_email="airflow@localhost",
        )
        notifier(email_context)
    except Exception:
        log.exception("Failed to send email notification")


def _execute_task(context: Context, ti: RuntimeTaskInstance, log: Logger):
    """Execute Task (optionally with a Timeout) and push Xcom results."""
    task = ti.task
    execute = task.execute

    if ti._ti_context_from_server and (next_method := ti._ti_context_from_server.next_method):
        from airflow.serialization.serialized_objects import BaseSerialization

        kwargs = BaseSerialization.deserialize(ti._ti_context_from_server.next_kwargs or {})

        execute = functools.partial(task.resume_execution, next_method=next_method, next_kwargs=kwargs)

    ctx = contextvars.copy_context()
    # Populate the context var so ExecutorSafeguard doesn't complain
    ctx.run(ExecutorSafeguard.tracker.set, task)

    # Export context in os.environ to make it available for operators to use.
    airflow_context_vars = context_to_airflow_vars(context, in_env_var_format=True)

Review comment (refactor):
The SmtpNotifier already has html template that does this.
I think we can simplify and reduce more of the code we have around creation of `default_html_content`
https://github.com/apache/airflow/pull/36226

Code after:
    try:
        from airflow.providers.smtp.notifications.smtp import SmtpNotifier
    except ImportError:
        log.error(
            "Failed to send task failure or retry email notification: "
            "`apache-airflow-providers-smtp` is not installed. "
            "Install this provider to enable email notifications."
        )
        return

    if not task.email:
        return

    subject_template_file = conf.get("email", "subject_template", fallback=None)

    # Read the template file if configured
    if subject_template_file and Path(subject_template_file).exists():
        subject = Path(subject_template_file).read_text()
    else:
        # Fallback to default
        subject = "Airflow alert: {{ti}}"

    html_content_template_file = conf.get("email", "html_content_template", fallback=None)

    # Read the template file if configured
    if html_content_template_file and Path(html_content_template_file).exists():
        html_content = Path(html_content_template_file).read_text()
    else:
        # Fallback to default
        # For reporting purposes, we report based on 1-indexed,
        # not 0-indexed lists (i.e. Try 1 instead of Try 0 for the first attempt).
        html_content = (
            "Try {{try_number}} out of {{max_tries + 1}}<br>"
            "Exception:<br>{{exception_html}}<br>"
            'Log: <a href="{{ti.log_url}}">Link</a><br>'
            "Host: {{ti.hostname}}<br>"
            'Mark success: <a href="{{ti.mark_success_url}}">Link</a><br>'
        )

    # Add exception_html to context for template rendering
    import html

    exception_html = html.escape(str(error)).replace("\n", "<br>")
    additional_context = {
        "exception": error,
        "exception_html": exception_html,
        "try_number": ti.try_number,
        "max_tries": ti.max_tries,
    }
    email_context = {**context, **additional_context}
    to_emails = task.email