Collectives™ on Stack Overflow

Find centralized, trusted content and collaborate around the technologies you use most.

Learn more about Collectives

Teams

Q&A for work

Connect and share knowledge within a single location that is structured and easy to search.

Learn more about Teams

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)
    

    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 1, 2020 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 1, 2020 at 15:25

    Thanks for contributing an answer to Stack Overflow!

    • Please be sure to answer the question. Provide details and share your research!

    But avoid

    • Asking for help, clarification, or responding to other answers.
    • Making statements based on opinion; back them up with references or personal experience.

    To learn more, see our tips on writing great answers.

  •