Note :

Ce tutoriel nécessite l'accès à Oracle Cloud. Pour vous inscrire à un compte gratuit, voir Démarrer avec le niveau gratuit d'Oracle Cloud Infrastructure.
Il utilise des exemples de valeurs pour les données d'identification, la location et les compartiments d'Oracle Cloud Infrastructure. À la fin de votre laboratoire, remplacez ces valeurs par celles qui sont propres à votre environnement en nuage.

Configurer les avis pour les tâches de flux de travail AIDP et les scénarios personnalisés à l'aide des avis OCI

Présentation

Ce tutoriel explique comment configurer des avis pour les tâches de flux de travail AIDP à l'aide du service d'avis et d'événements OCI. Vous apprendrez également à implémenter une logique de notification personnalisée à l'aide d'un programme Python dans un carnet AIDP.

Le service d'avis OCI utilise un modèle de publication-abonnement pour transmettre les messages de manière fiable et sécurisée.

Objectifs

À la fin de ce tutoriel, vous pourrez :

Créer un sujet d'avis OCI et un abonnement
Configurer les règles d'événement pour le statut de la tâche de flux de travail AIDP
Recevoir des avis par courriel pour les échecs d'emploi
Mettre en oeuvre une logique de notification personnalisée à l'aide de Python
Déclencher des avis par programmation

Conditions requises

Accéder à la console OCI
Espace de travail AIDP avec tâches de flux de travail
Autorisations pour créer des avis et des événements
Configuration de la configuration OCI pour l'utilisation de la trousse SDK Python

Tâche 1 : Créer un sujet d'avis et un abonnement

Connectez-vous à la console OCI.
Naviguez jusqu'à Services de développement → Avis.
Cliquez sur Créer un sujet et fournissez les informations suivantes :
- Nom
- Description
Créer un abonnement :
- Protocole : Courriel
- Courriel : Votre adresse de courriel
Vérifiez votre email et confirmez l'abonnement.

Note : 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 flux de travail AIDP

Naviguez jusqu'à Observabilité et gestion → Service d'événements → Règles.
Cliquez sur Créer une règle.
Configurer :
- Service : Lac de données intelligent
- Type d'événement : Exécuter la tâche - Fin
- Nom d'attribut 1 : jobKey
- Valeurs d'attribut1 : Conserver l'ID flux de travail helpp
- Nom d'attribut 2 : jobStatus
- Valeurs d'attribut 2 : Échec
Définir l'action :
- Sélectionner un sujet d'avis créé précédemment

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

Tâche 3 : Valider l'avis par courriel

Déclencher ou exécuter une tâche de flux des travaux.
Si nécessaire, forcez un scénario d'échec.
Vérifiez votre boîte de réception de courriel.

Note : L'avis inclut les détails de la tâche, tels que le nom et le statut de la tâche.

Tâche 4 : Mettre en oeuvre un avis personnalisé à l'aide de Python

Voici un exemple de programme Python pour envoyer des avis à l'aide de la trousse 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 d'avis personnalisé avancé

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 un avis personnalisé à partir du carnet AIDP

Vous pouvez appeler la logique de notification dans votre flux de travail ou votre carnet :

from notifier import notify_failure

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

Dépannage et conseils

Conseil : Assurez-vous que le chemin d'accès au fichier de configuration OCI est correct.

Note : Vérifiez l'OCID du sujet et le statut de l'abonnement.

Conseil : Utilisez les nouvelles tentatives de l'avis personnalisé pour des raisons de fiabilité.

Étapes suivantes

Ajouter des abonnements Slack ou HTTPS
Étendre les avis pour les événements de réussite
Intégrer à des tableaux de bord de surveillance

Remerciements

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

Ressources d'apprentissage supplémentaires

Explorez d'autres laboratoires sur le site docs.oracle.com/learn ou accédez à plus de contenu d'apprentissage gratuit sur la chaîne YouTube d'Oracle Learning. De plus, visitez education.oracle.com/learning-explorer pour devenir un explorateur Oracle Learning.

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

Titre et informations sur les droits d'auteur

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

G56826-01

Oracle et/ou ses sociétés affiliées.