応用例：「ログをメール通知やチャット通知（Slack/Teams）に連携する」
スケジューリングしたパイプラインに ログ通知機能 を追加すると、エラーや完了状況をリアルタイムで把握できます。メールやチャット（Slack/Teams）に連携すれば、チーム全体で監視・共有が可能になります。
メール通知の例（エラー発生時に送信）
import logging
import smtplib
from email.mime.text import MIMEText
logging.basicConfig(filename="pipeline.log", level=logging.INFO)
def notify_email(message):
sender = "your_email@example.com"
receiver = "receiver@example.com"
msg = MIMEText(message)
msg["Subject"] = "パイプライン通知"
msg["From"] = sender
msg["To"] = receiver
with smtplib.SMTP_SSL("smtp.gmail.com", 465) as server:
server.login("your_email@example.com", "your_password")
server.sendmail(sender, receiver, msg.as_string())
def pipeline():
try:
logging.info("パイプライン開始")
# 処理...
logging.info("パイプライン完了")
notify_email("パイプラインが正常に完了しました。")
except Exception as e:
logging.error(f"エラー発生: {e}", exc_info=True)
notify_email(f"エラー発生: {e}")
👉 エラーや完了時にメール通知が送られます。
Slack 通知の例（Webhook を利用）
Slack の Incoming Webhook URL を使って通知できます。
import requests
def notify_slack(message):
webhook_url = "https://hooks.slack.com/services/XXXX/XXXX/XXXX"
payload = {"text": message}
requests.post(webhook_url, json=payload)
def pipeline():
try:
# 処理...
notify_slack("✅ パイプラインが正常に完了しました。")
except Exception as e:
notify_slack(f"❌ エラー発生: {e}")
👉 Slack チャンネルにリアルタイムで通知が届きます。
Microsoft Teams 通知の例（Webhook を利用）
Teams でも Incoming Webhook を設定して通知可能です。
import requests
def notify_teams(message):
webhook_url = "https://outlook.office.com/webhook/XXXX/IncomingWebhook/XXXX"
payload = {"text": message}
requests.post(webhook_url, json=payload)
def pipeline():
try:
# 処理...
notify_teams("✅ パイプラインが正常に完了しました。")
except Exception as e:
notify_teams(f"❌ エラー発生: {e}")
👉 Teams チャネルに通知が届き、チーム全員で状況を確認できます。
解説ポイント
- メール通知 → 個人や少人数向けに便利。
- Slack/Teams通知 → チーム全体でリアルタイム監視に最適。
- Webhook連携 → シンプルに HTTP POST で通知可能。
- エラーハンドリングと組み合わせ → 成功/失敗を即座に共有できる。
応用の場面
- 定期レポート配布の監視 → 成功/失敗をチームに通知。
- 障害対応 → エラー発生を即座に共有して迅速対応。
- 業務効率化 → ログを自動通知して人手による確認を不要に。
まとめ
- ログを メール/Slack/Teams に通知すれば、リアルタイム監視が可能。
- Webhook を使えば簡単にチャット連携できる。
- チーム全体で状況を共有できるので、安定運用に役立つ。
