Skip links

Comment créer et personnaliser des notifications Teams et Slacks avec Python

Déclencher automatiquement une notification Slack ou Teams avec python dès qu'une nouvelle exportation des données brutes GA4 vers BigQuery est disponible.

Déclencher automatiquement une notification Slack ou Teams avec python dès qu’une nouvelle exportation des données brutes GA4 vers BigQuery est disponible.

Chez Hanalytics, nous avons mis en place des processus de monitoring de jobs d’automatisation des flux de rapports avec Slack ou Teams. Nous faisons en sorte d’être notifié à chaque étape de toutes nos Data Pipeline. Ces processus nous permettent de rapidement identifier les jobs tombés en erreur, de nous assurer de la santé de toutes les pipelines et de garantir la délivrabilité des metrics et la mise à jour des reportings à nos clients. 

Créer une alerte Teams ou Slack est assez simple et efficace, il suffit d’un script python de quelques lignes pour envoyer une notifications via un Webhook.

Cas Pratique Slack

Nous souhaitons déclencher avertir l’ensemble des équipes Data dès que l’exportation des données GA4 de la veille est disponible dans BigQuery. Pour rappel, il n’y a pas une heure fixe d’exportation des données vers BigQuery. Les données peuvent être exportées lundi à 10h, mardi à 15h ou Mercredi à 8h30. Les équipes restent dépendantes de la disponibilité de ces données là pour pouvoir lancer les différents traitements de consolidation de données et de mise à jour des reportings. 

Pour détecter la disponibilité des données, il suffit de suivre le tuto décrit dans cet article. Il faut mettre en place l’architecture suivante : 

  • Détecter la disponibilité des données via Cloud Logging et créer une Pub/Sub qui va déclencher une Cloud Function. 
  • Créer le script d’envoie de la notification de la dispo des données vers Slack

source : data-hanalytics.io

Le script à intégré dans la Cloud Function est le suivant : 

 

				
					from datetime import date
today = date.today()
# Textual month, day and year	
date_today = today.strftime("%B %d, %Y")
print(date_today)

import requests
import json 
def send_slack_message(payload, slack_webhook):
    """Send a Slack message to a channel via a webhook. 
    
    Args:
        payload (dict): Dictionary containing Slack message, i.e. {"text": "This is a test"}
        webhook (str): Full Slack webhook URL for your chosen channel. 
    
    Returns:
        HTTP response code, i.e. <Response [503]>
    """

    return requests.post(slack_webhook, json.dumps(payload))


slack_webhook = ["WEBHOOK_URL"]
payload = {
	"blocks": [
{
			"type": "header",
			"text": {
				"type": "plain_text",
				"text": ":loud_sound:  Hanalytics Custom Alert  :loud_sound:"
			}
		},
		{
			"type": "context",
			"elements": [
				{
					"text": date_today + " | Google Analytics 4 Alerting",
					"type": "mrkdwn"
				}
			]
		},
		{
			"type": "divider"
		},
		{
			"type": "section",
			"text": {
				"type": "mrkdwn",
				"text": "Hello la team , les données GA4 J-1 sont dispos dans BigQuery dans le Dataset Analytics - XXXXXXXXXXX"
			},
			"accessory": {
				"type": "button",
				"text": {
					"type": "plain_text",
					"text": "Accès à BigQuery"				
          },
				"value": "click_me_123",
				"url": ["BIGQUERY_URL"],
				"action_id": "button-action"          

			}
		}
	]
}

def ga4_data_update(event, context):
    """Triggered from a message on a Cloud Pub/Sub topic.
    Args:
         event (dict): Event payload.
         context (google.cloud.functions.Context): Metadata for the event.
    """
    send_slack_message(payload, slack_webhook)
    print("Alert Sent to Slack")

				
			

Une fois que le flux est mis en place et que la Cloud Function est disponible, les notifications sont envoyées automatiquement en quasi temps réel dès qu’une nouvelle table des données brutes GA4 est disponible dans le dataset BigQuery dédié à votre compte Google Analytics. 

Création de la table dans BigQuery le 26/05 à 10:24 -> 

La notification Slack : 

Notez que pour que envoyer une notification , il faut tout d’abord créer une application Slack

Cas Pratique Teams

Pour envoyer une notification dans Teams, la procédure est tout aussi simple : 


				
					import requests
import json 

# Send message to teams 
webhook_url = 'INSERT YOUR WEBHOOK URL'
def send_teams(webhook_url:str, content:str, title:str) -> int:
    """
      - Send a teams notification to the desired webhook_url
      - Returns the status code of the HTTP request
        - webhook_url : the url you got from the teams webhook configuration
        - content : your formatted notification content
        - title : the message that'll be displayed as title, and on phone notifications
        - color (optional) : hexadecimal code of the notification's top line color, default corresponds to black
    """
    response = requests.post(
        url=webhook_url,
        headers={"Content-Type": "application/json"},
        json={
            "summary": title,
            "sections": [{
                "activityTitle": title,
                "activitySubtitle": content
            }],
        },
    )
    return response.status_code # Should be 200

title="Hanalytics Data Monitoring bot 🤖"
content = """
Hello la team , les données GA4 J-1 sont dispos dans BigQuery dans le Dataset Analytics - XXXXXXXXXXX 
<br>
🚀🚀🚀🚀
"""
send_teams(webhook_url, content, title)
				
			

La notification est envoyée automatiquement en quasi temps réel à Teams -> 


Leave a comment

Explore
Drag