Remarques :

Ce tutoriel nécessite l'accès à Oracle Cloud. Pour vous inscrire à un compte gratuit, reportez-vous à Introduction à Oracle Cloud Infrastructure Free Tier.
Il utilise des exemples de valeur pour les informations d'identification, la location et les compartiments Oracle Cloud Infrastructure. Lorsque vous terminez votre atelier, remplacez ces valeurs par des valeurs propres à votre environnement cloud.

Configuration des notifications pour les travaux de workflow AIDP et les scénarios personnalisés à l'aide des notifications OCI

Introduction

Ce tutoriel explique comment configurer des notifications pour les travaux de workflow AIDP à l'aide du service OCI Notifications and Events. Vous apprendrez également à implémenter une logique de notification personnalisée à l'aide d'un programme Python dans un bloc-notes AIDP.

Le service OCI Notifications utilise un modèle de publication/souscription pour diffuser des messages de manière fiable et sécurisée.

Objectifs

A la fin de ce tutoriel, vous pourrez :

Créer un sujet et un abonnement OCI Notification
Configurer les règles d'événement pour le statut du travail de workflow AIDP
Recevoir des notifications par courriel pour les échecs de travail
Implémenter une logique de notification personnalisée à l'aide de Python
Déclencher des notifications par programmation

Prérequis

Accès à la console OCI
Espace de travail AIDP avec travaux de workflow
Autorisations permettant de créer des notifications et des événements
Configuration OCI pour l'utilisation du kit SDK Python

Tâche 1 : créer un sujet de notification et un abonnement

Connectez-vous à OCI Console.
Accédez à Services de développeur → Notifications.
Cliquez sur Créer un sujet et indiquez les informations suivantes :
- Nom
- Description
Créez un abonnement :
- Protocole : courriel
- Adresse électronique : votre adresse électronique
Vérifiez votre adresse électronique et confirmez l'abonnement.

Remarque : le statut de l'abonnement passe de En attente à Actif après confirmation.

Tâche 2 : configurer la règle d'événement pour le workflow AIDP

Accédez à Observation & gestion → Service d'événements → Règles.
Cliquez sur Créer une règle,
Configurez les éléments suivants :
- Service : lac de données intelligent
- Type d'événement : Exécuter le travail - Fin
- Nom d'attribut1 : jobKey
- Valeurs d'attribut1 : conserver l'ID workflow aidep
- Nom d'attribut 2 : jobStatus
- Valeurs d'attribut2 : Echec
Action définie:
- Sélectionner une rubrique de notification créée précédemment

A savoir : Vous pouvez personnaliser les règles pour d'autres statuts tels que Succès ou En cours d'exécution.

Tâche 3 : Valider la notification par e-mail

Déclencher ou exécuter un travail de workflow.
Forcer un scénario d'échec si nécessaire.
Vérifiez votre boîte de réception.

Remarque : La notification inclut des détails de travail tels que le nom et le statut du travail.

Tâche 4 : Implémenter une notification personnalisée à l'aide de Python

Voici un exemple de programme Python permettant d'envoyer des notifications à l'aide du kit SDK OCI.

import oci
from oci.ons.models import MessageDetails

config = oci.config.from_file("~/.oci/config", "DEFAULT")
client = oci.ons.NotificationDataPlaneClient(config)

topic_id = "<your_topic_ocid>"

message_details = MessageDetails(
    title="Test Notification",
    body="Hello from AIDP!"
)

response = client.publish_message(
    topic_id=topic_id,
    message_details=message_details
)

print("Message sent:", response.data.message_id)

Tâche 5 : Classe de notification personnalisée avancée

import oci
import logging
import time
from typing import List, Optional, Dict

logger = logging.getLogger("OCI_Notifier")
logger.setLevel(logging.INFO)

handler = logging.StreamHandler()
formatter = logging.Formatter(
    "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
handler.setFormatter(formatter)
logger.addHandler(handler)

class OCINotifier:
    def __init__(
        self,
        topic_id: str,
        config_file: str = "~/.oci/config",
        profile: str = "DEFAULT",
        max_retries: int = 3,
        retry_delay: int = 2,
    ):
        """
        Initialize OCI Notification client

        Args:
            topic_id: OCI Notification Topic OCID
            config_file: Path to OCI config
            profile: Config profile name
            max_retries: Retry attempts
            retry_delay: Delay between retries (seconds)
        """
        self.topic_id = topic_id
        self.max_retries = max_retries
        self.retry_delay = retry_delay

        try:
            self.config = oci.config.from_file(config_file, profile)
            self.client = oci.ons.NotificationDataPlaneClient(self.config)
            logger.info("OCI Notifier initialized successfully")
        except Exception as e:
            logger.error(f"Failed to initialize OCI client: {e}")
            raise

    
    def send(
        self,
        title: str,
        message: str,
        metadata: Optional[Dict] = None,
    ) -> bool:
        """
        Send notification message

        Args:
            title: Notification subject
            message: Notification body
            metadata: Optional metadata dictionary

        Returns:
            bool: True if success, False otherwise
        """

        payload = {
            "title": title,
            "body": message,
        }

        if metadata:
            payload["metadata"] = metadata

        for attempt in range(1, self.max_retries + 1):
            try:
                response = self.client.publish_message(
                    self.topic_id,
                    payload
                )
                logger.info(
                    f"Notification sent successfully | "
                    f"Message ID: {response.data.message_id}"
                )
                return True

            except Exception as e:
                logger.error(
                    f"Attempt {attempt} failed: {str(e)}"
                )
                if attempt < self.max_retries:
                    time.sleep(self.retry_delay)
                else:
                    logger.error("Max retries reached. Notification failed.")
                    return False



def notify_success(notifier: OCINotifier, job_name: str):
    notifier.send(
        title=f"{job_name} SUCCESS",
        message=f"Job '{job_name}' completed successfully."
    )


def notify_failure(notifier: OCINotifier, job_name: str, error: str):
    notifier.send(
        title=f"{job_name} FAILED",
        message=f"Job '{job_name}' failed.\nError: {error}"
    )

Tâche 6 : appeler une notification personnalisée à partir d'un bloc-notes AIDP

Vous pouvez appeler la logique de notification dans votre workflow ou bloc-notes :

from notifier import notify_failure

notify_failure(notifier, "SampleJob", "Error details here")

Dépannage et conseils

A savoir : Assurez-vous que le chemin du fichier de configuration OCI est correct.

Remarque : vérifiez l'OCID du sujet et le statut de l'abonnement.

A savoir : Utilisez les nouvelles tentatives dans le notifiant personnalisé pour plus de fiabilité.

Etapes suivantes

Ajouter des abonnements Slack ou HTTPS
Prolonger les notifications pour les événements de succès
Intégration aux tableaux de bord de surveillance

Remerciements

Auteur - Pavan Upadhyay (ingénieur cloud principal), Saket Bihari (ingénieur cloud principal)

Ressources de formation supplémentaires

Explorez d'autres ateliers sur docs.oracle.com/learn ou accédez à d'autres contenus d'apprentissage gratuits sur la chaîne YouTube Oracle Learning. En outre, visitez le site education.oracle.com/learning-explorer pour devenir un explorateur Oracle Learning.

Pour obtenir de la documentation sur le produit, consultez Oracle Help Center.

Informations relatives au titre et au copyright

Configure Notifications for AIDP Workflow Jobs and Custom Scenarios Using OCI Notifications

G56827-01

Oracle et/ou ses affiliés.