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!:

1. 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

2. 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=['[email protected]'],
           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=['[email protected]'],
        subject=email_subject,
        files=[export_file_path,],
        html_content='<br>')
    op.execute(kwargs)

Answer 1

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=['[email protected]'],
        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)