[Airflow] Python/email Operator간 Xcom 사용

---

Email Operator에서 사용 가능한 Template

▪ 'to', 'subject', 'html_content', 'files' 파라미터에 Template 적용 가능 

Python Operator에서 Email Operator로 Xcom 전달하기

▪ PythonOperator에서 return한 값을 ti객체에 xcom_pull 함수를 사용하여 값 받아오기

▪ 이때 template 문법을 사용해야함 

---

dags_python_email_xcom.py

▪ 'something_task'에서 'Success' 와 'Fail' 중 하나의 값을 랜덤으로 return해서 Xcom 전달

▪ 'send_email'에서 return한 값을 가져오고, 해당 시간을 기준으로 이메일 제목과 내용을 작성해서 전송 

 

from airflow import DAG
import pendulum
import datetime
from airflow.decorators import task
from airflow.operators.email import EmailOperator
 
with DAG(
    dag_id="dags_python_email_operator",
    schedule="0 8 1 * *",
    start_date=pendulum.datetime(2023, 12, 19, tz="Asia/Seoul"),
    catchup=False
) as dag:
    
    @task(task_id='something_task')
    def some_logic(**kwargs):
        from random import choice
        return choice(['Success','Fail'])
    
    send_email = EmailOperator(
        task_id='send_email',
        to='dojun43@gmail.com',
        subject='{{ data_interval_end.in_timezone("Asia/Seoul") | ds}} some_logic 처리결과',
        html_content='{{ data_interval_end.in_timezone("Asia/Seoul") | ds}} 처리 결과는 <br> \
                    {{ ti.xcom_pull(task_ids="something_task")}} 했습니다. <br>'
    )
 
    some_logic() >> send_email

 

Dag 실행하기

▪ dag을 airflow에 upload하고, webserver 접속해서 dag unpause 

▪ 이메일을 확인해보면 정상적으로 이메일이 전송됨

 

 

 

 

참고:

Airflow 마스터 클래스 강의 - 인프런