Airflow 2.4.2结合SendGrid发送带template_id模板邮件失败求助

我正在使用SendGrid与Apache Airflow（2.4.2版本）发送邮件，目前可正常发送普通邮件，但尝试传入template_id使用模板发送时，邮件未加载模板样式。看起来我传递template_id的方式有误，请问如何正确发送带模板的邮件？

附带代码如下：

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.providers.sendgrid.utils.emailer import send_email

from airflow.operators.email import EmailOperator

from datetime import datetime


default_args = {
    'owner': 'your_name',
    'start_date': datetime(2023, 10, 16, 15, 35, 0),
    'retries': 1,
}

#Task to send an email using SendGrid
def _send_email():

    postgres_hook = PostgresHook(postgres_conn_id="postgres")
    sql_query = "select email from Users where firstname='Raghav';"
    results = postgres_hook.get_records(sql_query)

    for row in results:
        email = row
        email_content = f"Hello  Email: {email}\n"

        send_email(
            to=email,
            subject='Test Email Subject',
            html_content=email_content,
            files=None,  # List of file paths to attach to the email
            cc=None,  # List of email addresses to CC
            bcc=None,  # List of email addresses to BCC
            conn_id='sendgrid_default',
            template_id='d-8cefbe24e73842d7810550fc441aceb0', #didn't work
            kwargs={
                'template_id':'d-8cefbe24e99942d7810550fc441aceb0'
            }
        )

with DAG('user_processing',
        start_date=datetime(2022,1,1),
        schedule_interval='@daily',
        catchup=False) as dag:

    email_user = PythonOperator(
        task_id = 'email_user',
        python_callable= _send_email
    )

    email_user

核心问题分析 #

1. Airflow的SendGrid版send_email不支持将template_id作为顶层参数传入，必须通过kwargs传递给底层SendGrid API。
2. 同时指定html_content会覆盖模板内容，SendGrid在收到自定义HTML时会忽略模板设置。
3. 从数据库获取的email是元组类型，需要提取出实际的邮箱字符串才能正确发送。

修正后的代码 #

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.providers.sendgrid.utils.emailer import send_email

from datetime import datetime


default_args = {
    'owner': 'your_name',
    'start_date': datetime(2023, 10, 16, 15, 35, 0),
    'retries': 1,
}

#Task to send an email using SendGrid template
def _send_email():
    postgres_hook = PostgresHook(postgres_conn_id="postgres")
    sql_query = "select email from Users where firstname='Raghav';"
    results = postgres_hook.get_records(sql_query)

    for row in results:
        # 提取元组中的邮箱字符串
        email = row[0]
        # 模板需要的动态数据（根据你的模板变量调整）
        template_data = {
            "username": "Raghav",
            "email": email
        }

        send_email(
            to=email,
            subject='Test Email Subject',
            # 使用模板时不需要指定html_content
            files=None,
            cc=None,
            bcc=None,
            conn_id='sendgrid_default',
            # 所有SendGrid API需要的参数都放在kwargs里
            kwargs={
                'template_id': 'd-8cefbe24e99942d7810550fc441aceb0',
                'dynamic_template_data': template_data
            }
        )

with DAG('user_processing',
        start_date=datetime(2022,1,1),
        schedule_interval='@daily',
        catchup=False) as dag:

    email_user = PythonOperator(
        task_id='email_user',
        python_callable=_send_email
    )

    email_user

关键修改点 #

• 移除html_content参数，确保SendGrid优先使用模板内容。
• 将template_id放入kwargs字典中，这是传递给SendGrid API的正确方式。
• 添加dynamic_template_data（按需调整），用于填充模板中的动态占位符。
• 修正email变量的提取逻辑，从数据库返回的元组中取出实际邮箱地址。

内容的提问来源于stack exchange，提问作者James Hameson