Apache Airflow 2.0.0.b2 - Dynamic EmailOperator [files] property

Question

TL;DR How do I create a dynamic EmailOperator that sends a file from a filepath that is an XCom property

Hello Folks,

I'm using Apache Airflow 2.0.0.b2. My issue is that my DAG creates a file whose name changes at runtime. I'd like to email this file out but I'm having problems getting the dynamic file name into my EmailOperator.

Things I've Tried that Have Failed!:

Use templating for files property.
```
files=["{{ ti.xcom_pull(key='OUTPUT_CSV') }}"],
```
Unfortunately, templating only works when the field in the operator is marked as such. files is not a templatable field on the EmailOperator

Dynamically create my task using a function

 def get_email_operator(?...):
    export_file_path = ti.xcom_pull(key='OUTPUT_CSV')
    email_subject = 'Some Subject'
    return EmailOperator(
        task_id="get_email_operator",
        to=['someemail@somedomain.net'],
        subject=email_subject,
        files=[export_file_path,],
        html_content='<br>',
        dag=current_dag)

..task3 >> get_email_operator() >> task4

Unfortunately, I can't seem to figure out how to pass the current **kwargs or ti information into my function call to get the current filepath.

EDIT: Elad's answer below set me in the right direction. The only thing I had to to do to make it work was to add kwargs when calling op.execute()

Solution:

def get_email_operator(**kwargs):
    export_file_path = kwargs['ti'].xcom_pull(key='OUTPUT_CSV')
    email_subject = 'Termed Drivers - ' + date_string
    op = EmailOperator(
        task_id="get_email_operator",
        to=['someemail@somedomain.net'],
        subject=email_subject,
        files=[export_file_path,],
        html_content='<br>')
    op.execute(kwargs)

Elad Kalif · Accepted Answer · 2022-07-24T08:02:38.653

files will be templated in Airflow 2 as the PR was merged last week.

However you don't need to wait for this you can wrap the current operator with your own custom operator specifying the list of templated fields.

like:

class MyEmailOperator(EmailOperator):
     template_fields = ('to', 'subject', 'html_content', 'files')

Then you can use MyEmailOperator in your code. files will be templated.

Another option is PythonOperator that wraps EmailOperator:

def get_email_operator(**context):
    xcom = context['ti'].xcom_pull(task_ids='OUTPUT_CSV')
    email_subject = 'Some Subject'
    op = EmailOperator(
        task_id="get_email_operator",
        to=['someemail@somedomain.net'],
        subject=email_subject,
        files=[xcom,],
        html_content='<br>')
    op.execute(context)

python = PythonOperator(
    task_id='archive_s3_file',
    dag=dag,
    python_callable=get_email_operator,
    provide_context=True
)

..task3 >> python >> task4

Edit:

Note: Using operator inside operator is not a good practice. You can read more about why in the following answer. If you are going to use the 2nd approach then a better idea would be to use send_email function directly (EmailOperator invokes this function)

Thanks a bunch. I went with the second solution because of fears about mis-architecting my pipelines with such little experience. Worked like a charm once I passed my context to the EmailOperator's execute method. — Black Dynamite, Dec 01 '20 at 15:07
@BlackDynamite NP. I advise you to at least try the first solution. It's much simpler and when the fix will be released in next versions you can easily remove the custom operator and replace it with the Airflow one. — Elad Kalif, Dec 01 '20 at 15:25

Apache Airflow 2.0.0.b2 - Dynamic EmailOperator [files] property

1 Answers1