Observação:

• Este tutorial requer acesso ao Oracle Cloud. Para se inscrever em uma conta gratuita, consulte Conceitos Básicos do Oracle Cloud Infrastructure Free Tier.
• Ele usa valores de exemplo para credenciais, tenancy e compartimentos do Oracle Cloud Infrastructure. Ao concluir seu laboratório, substitua esses valores por valores específicos do seu ambiente de nuvem.

Configurar Notificações para Jobs de Workflow AIDP e Cenários Personalizados Usando Notificações do OCI

Introdução

Este tutorial explica como configurar notificações para jobs de workflow do AIDP usando o serviço OCI Notifications e Events. Você também aprenderá a implementar lógica de notificação personalizada usando um programa Python em um notebook AIDP.

O serviço OCI Notifications usa um modelo de publicação-assinatura para entregar mensagens de forma confiável e segura.

Objetivos

Ao final deste tutorial, você será capaz de:

• Criar um tópico e uma assinatura do OCI Notification
• Configurar regras de evento para status do job de workflow AIDP
• Receber notificações por e-mail sobre falhas de job
• Implementar lógica de notificação personalizada usando Python
• Acionar notificações programaticamente

Pré-requisitos

• Acesso à Console do OCI
• Espaço de trabalho AIDP com jobs de workflow
• Permissões para criar Notificações e Eventos
• Configuração do OCI para uso do Python SDK

---

Tarefa 1: Criar Tópico de Notificação e Assinatura

1. Faça log-in na Console da OCI.
2. Navegue até Serviços do Desenvolvedor → Notificações.
3. Clique em Criar Tópico e forneça:
   • Nome
   • Descrição

   

4. Criar uma assinatura:
   • Protocolo: E-mail
   • E-mail: Seu endereço de e-mail

   

5. Verifique seu email e confirme a inscrição.

   

Observação: o status da assinatura muda de Pendente para Ativo após a confirmação.

---

Tarefa 2: Configurar regra de evento para fluxo de trabalho AIDP

1. Navegue até Observability & Management → Events Service → Rules.
2. Clique em Criar Regra.
3. Configurar:
   • Serviço: Intelligent Data Lake
   • Tipo de Evento: Executar Job - Final
   • Nome1 do Atributo: jobKey
   • Valores de atributo1: Manter id do workflow do ajudp
   • Nome do Atributo 2 : jobStatus
   • Valores de atributo2: Falha
4. Definir Ação:
   • Selecionar Tópico de Notificação criado anteriormente

   

Dica: você pode personalizar regras para outros status como Sucesso ou Em Execução.

---

Tarefa 3: Validar Notificação por E-mail

1. Acionar ou executar um job de workflow.
2. Forçar um cenário de falha, se necessário.

3. Verifique sua caixa de entrada de e-mail.

   

Observação: a notificação inclui detalhes do job, como nome e status do job.

---

Tarefa 4: Implementar Notificação Personalizada Usando o Python

Veja a seguir um programa Python de amostra para enviar notificações usando o OCI SDK.

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)

---

Tarefa 5: Classe de Notificação Personalizada Avançada

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}"
    )

---

Tarefa 6: Chamar Notificação Personalizada do Notebook AIDP

Você pode chamar a lógica de notificação dentro do seu workflow ou notebook:

from notifier import notify_failure

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

---

Diagnosticando e Solucionando Problemas

Dica: Certifique-se de que o caminho do arquivo de configuração do OCI esteja correto.

Observação: Verifique o OCID do tópico e o status da assinatura.

Dica: Use repetições no notificador personalizado para confiabilidade.

---

Próximas Etapas

• Adicionar assinaturas do Slack ou HTTPS
• Estender notificações para eventos de sucesso
• Integre com painéis de monitoramento

---

Agradecimentos

• Autor - Pavan Upadhyay (Engenheiro de Nuvem Principal), Saket Bihari (Engenheiro de Nuvem Principal)

Mais Recursos de Aprendizado

Explore outros laboratórios em docs.oracle.com/learn ou acesse mais conteúdo de aprendizado gratuito no canal do Oracle Learning no YouTube. Além disso, acesse education.oracle.com/learning-explorer para se tornar um Oracle Learning Explorer.

Para obter a documentação do produto, visite o Oracle Help Center.

---

Título e Informações de Copyright

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

G56831-01

Copyright ©2026,

Oracle e/ou suas empresas afiliadas.