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