formatted the script

Author: Chideraozigbo

dags/notification/email.py

@@ -25,7 +25,7 @@ def notification_email(context, state):
     email_port = int(Variable.get("email_port"))
 
     print(context)
-    subject = f"Airflow Alert: Task '{task}' in DAG '{dag}' has {state.upper()}"
+    subject = f"Airflow Alert: Task in DAG '{dag}' has {state.upper()}"
     body = f"""
     Hey {dag_owner}
 
@@ -42,7 +42,9 @@ def notification_email(context, state):
 
     ssl_context = ssl.create_default_context()
 
-    with smtplib.SMTP_SSL(mail_server, email_port, context=ssl_context) as smtp:
+    with smtplib.SMTP_SSL(
+        mail_server, email_port, context=ssl_context
+    ) as smtp:
         smtp.login(email_sender, email_password)
         smtp.sendmail(email_sender, email_receiver, em.as_string())
         print("Email Sent Successfully")
@@ -53,5 +55,5 @@ def task_state_alert(context):
     This function send notification of a dags task
     """
     state = context.get("task_instance").state
-    if state in ("success","failed"):
+    if state in ("success", "failed"):
         notification_email(context, state)

dags/test.py

@@ -11,8 +11,8 @@
     "email_on_failure": True,
     "email_on_retry": False,
     "email_on_success": True,
-    'on_failure_callback': task_state_alert,
-    'on_success_callback': task_state_alert,
+    "on_failure_callback": task_state_alert,
+    "on_success_callback": task_state_alert,
     "email": ["chideraozigbo@gmail.com"],
     "retries": 1,
     "retry_delay": timedelta(seconds=3),