Airflow-MS-Teams-Operator
Airflow-MS-Teams-Operator copied to clipboard
Button is not showing up
Button is not showing up in the notification

from ms_teams_webhook_operator import MSTeamsWebhookOperator
def on_failure(context):
dag_id = context['dag_run'].dag_id
task_id = context['task_instance'].task_id
context['task_instance'].xcom_push(key=dag_id, value=True)
logs_url = "file:///Users/shubamsachdeva/airflow/logs/{}/{}/{}".format(
dag_id, task_id, context['ts'])
teams_notification = MSTeamsWebhookOperator(
task_id="msteams_notify_failure",
trigger_rule="all_done",
#message="`{}` has failed on task: `{}`. View Logs at: `{}`".format(dag_id, task_id, logs_url),
message = "**Hello from Airflow!**",
subtitle = "`{}` has failed on task: `{}`".format(dag_id, task_id),
button_text="View log",
button_url=logs_url,
theme_color="FF0000",
http_conn_id="msteams-webhook-url")
teams_notification.execute(context)
print('Error Notification is sent to your MS Team Channel')
If I had to guess it's your logs_url which appears to be a security risk, it's not a good idea to use file:/// URLs and I wonder if the webhook is filtering that out (that's just a guess). Try replacing it with an https:// URL. See my example here
If you want to prove or disprove the point you can try with hardcoded values first, before string formatting.
from ms_teams_webhook_operator import MSTeamsWebhookOperator
def on_failure(context):
dag_id = context['dag_run'].dag_id
task_id = context['task_instance'].task_id
context['task_instance'].xcom_push(key=dag_id, value=True)
logs_url = "https://example.com"
teams_notification = MSTeamsWebhookOperator(
task_id="msteams_notify_failure",
trigger_rule="all_done",
message = "**Hello from Airflow!**",
subtitle = "My subtitle",
button_text="View log",
button_url=logs_url,
theme_color="FF0000",
http_conn_id="msteams-webhook-url")
teams_notification.execute(context)
print('Error Notification is sent to your MS Team Channel')