How to trigger email for airflow task run

user_9079461 0
Viewed 295

We have an adhoc airflow DAG, which anyone can trigger to run manually from team of 50+.

We can check airflow audit logs who triggered the DAG via dag id and we can also get email upon failure.

But we are more curious to know if we can get email upon DAG start OR at the start of each task run, this will help us understand and track activity and usage/command executed from adhoc DAG.

airflow
1 Answers

@Khilesh Chauhan

There are a number of ways to achieve your intended outcome, in no particular order:

1. Task or Dag Level Callbacks Official Callback Reference

We can take advantage of the on_success_callback callback, which can be harnessed in two distinct places.

# use a function inside a specific PythonOperator, for task level control
task = PythonOperator(
  task_id='your_task',
  on_success_callback=send_mail,
)

# use it inside your DAG initiation 
dag = DAG(
    dag_id='your_task',
    on_failure_callback=send_mail
)

We can write an example send_mail function, which leverages the send_email utility.

from airflow.utils.email import send_email

def send_mail(**context):
  task = context['task_instance'].task

  subject = f'Airflow task has successfully completed {task.task_id}'
  body = f'Hi, this is an alert to let you know that your task {task.task_id} has completed successfully.'

  send_email(
    dag.default_args['email'],
    subject,
    body
  )

2. Add an EmailOperator to your DAG Official Email Operator Reference

You could add an EmailOperator task at the beginning of your DAG.

from airflow.operators.email_operator import EmailOperator

email = EmailOperator(
        task_id='alert_DAG_start',
        to='your@email.come',
        subject='DAG Initiated - start {{ ds }}',
        html_content=""" <h1>Some Content</h1> """
)

3. Create a function that uses a PythonOperator that executes send_email

You might need more control, such as including logging info etc. So you might want more control to use a PythonOperator.

I hope this helps you to resolve your problem.

Update

To answer your second question to get the username, I have created a function for you to use. We can import the session context manager, then use the .query method, for your debugging purposes, I loop through the array. you can see the username at index 3.

from airflow.models.log import Log
from airflow.utils.db import create_session

def return_user_name(**context):
  """
  return the username for the executed tasks
  """

  dag_id = context['task_instance'].dag_id

  with create_session() as session:
   result = session.query(Log.dttm, Log.dag_id, Log.execution_date, Log.owner, Log.extra).filter(Log.dag_id == dag_id, Log.event == 'trigger').first()
   for index, result in enumerate(result):
     print(index, result)
user_122793260
Related