使用 Airflow 和 Sendgrid 发送包含 template_id 的电子邮件时出现问题

Question

我有一个问题。我正在尝试使用 Sendgrid 和气流（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

提前致谢

Answer 1

这里是 Twilio 开发倡导者。

要使用 SendGrid 模板发送电子邮件，您应该使用 SendGridEmailOperator 而不是

send_email

模块中的

airflow.providers.sendgrid.utils.emailer

函数。以下是如何修改代码以使用 SendGrid 模板发送电子邮件：

from airflow import DAG
from airflow.providers.sendgrid.operators.sendgrid_email import SendGridEmailOperator
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():
    # Your code to retrieve the recipient email address and any other necessary data

    sendgrid_email = SendGridEmailOperator(
        task_id='sendgrid_email_task',
        to='[email protected]',
        subject='Test Email Subject',
        html_content='Hello Email: [email protected]',
        conn_id='sendgrid_default',
        template_id='d-8cefbe24e73842d7810550fc441aceb0',  # Your template ID
    )
    sendgrid_email.execute(context=None)

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

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

    email_user

在上面的代码中，我们从

airflow.providers.sendgrid.operators.sendgrid_email

模块导入 SendGridEmailOperator，该模块专门设计用于与 SendGrid 模板配合使用。然后，我们使用此运算符发送带有模板的电子邮件，并指定模板 ID，就像您在代码中所做的那样。

确保调整收件人的电子邮件地址以及电子邮件内容中需要包含的任何其他数据。此代码应使用指定的 SendGrid 模板发送电子邮件。

我希望这有帮助！