AirflowのジョブでTeamsに通知する方法

前回記事でOCIのPythonSDKを実行する紹介しました。
今回はその続きでPythonSDKを実行した結果をTeamsに通知する方法をご紹介します。

TeamsでWebhookを設定する

コネクタへアクセス

通知を送りたいチャンネルから「コネクタ」をクリック

Webhookの追加

「webhook」と検索し、追加すると「構成」に変わるので、構成をクリックします。

Webhook名を入力

ここでは「test」という適当な名前にし「作成」をクリックします。

WebhookのURLを確認する

問題なくwebookを作成するとURLが表示されるので、コピーしておく

Curlでテスト実行してみる

curl -X POST -d '{"text":"foo"}' https://******.webhook.office.com/webhookb2/*****d53219

下記のような感じで通知されます。

PythonOperatorからCurlを実行する

下記コードの"hogehoge"が通知時の文言になりますので、通知したい内容にカスタマイズすることで通知できます。
ちなみに、CurlコマンドをPythonに変換してくれるサイトがあるので利用させていただきました。
https://curlconverter.com/

    def curl2teams(**kwargs):
        cost = object_storage_cost()
        headers = {
            'Content-Type': 'application/x-www-form-urlencoded',
        }
        data = '{"text":"hogehoge"}'
        response = requests.post('https://********.webhook.office.com/webhookb2/e545****d53219', headers=headers, data=data)
	
    Curl2Teams = PythonOperator(
        task_id='curl2teams',
        python_callable=curl2teams,
    )