我創建了一個自定義電子郵件警報 python 檔案，我在另一個 dag 中呼叫該函式來發送電子郵件。可以說 - 電子郵件警報檔案是custom_alert.py
from airflow.operators.email import EmailOperator
from airflow.utils.email import send_email
def custom_failure_email(context, **kwargs):
"""Send custom email alerts."""
dag_run = context.get('task_instance').dag_id
subject = f"[ActionReq]-dag failure-{dag_run}"
# email contents
body= """Hi Team,<br><br>
<b style="font-size:15px;color:red;">Airflow job on error, please find details below.</b>
Thank you!,<br>
)
email_list = ['[email protected]', '[email protected]']
for i in range(len(email_list)):
send_email(str(email_list[i]),subject,body)
在父 DAG 中：可以說**email.py**- 我正在呼叫上述函式來發送失敗電子郵件。
from custom_alert import custom_failure_email
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'email': ['[email protected]'],
'email_on_failure': False,
'on_failure_callback': custom_failure_email}
這樣我就可以發送自定義電子郵件，但我的收件人串列對于每個 dag 都是相同的。請告訴我，我如何自定義它以針對不同的 dag 發送不同的收件人電子郵件地址。我如何從 Parent Dag 傳遞收件人電子郵件地址。?
uj5u.com熱心網友回復：
一個選項是在您的然后從背景關系中檢索此值中設定paramsdict 。DAG
def custom_failure_email(context):
"""Send custom email alerts."""
#...
email_list = context['dag'].params['mailing_list']
for i in range(len(email_list)):
send_email(str(email_list[i]),subject,body)
default_args = {
#....
'params': {
'mailing_list': ['[email protected]', '[email protected]']
},
'email_on_failure': False,
'on_failure_callback': custom_failure_email
}
uj5u.com熱心網友回復：
還有另一種方法可以使用 functools 來實作這一點。
from functools import partial
new_custom_failure_email = partial(custom_failure_email,
email_list=['[email protected]'])
# Now Pass this in default _args
'on_failure_callback': new_custom_failure_email
我們需要將email_list作為引數傳遞給**custom_alert.py**
轉載請註明出處，本文鏈接：https://www.uj5u.com/caozuo/496260.html