Отправка сообщения в Microsoft Teams при завершении задания

Обновлен

Webhooks

Advanced

Menu

Введение

Это руководство покажет вам, как настроить интеграцию между заданиями dbt Cloud и Microsoft Teams с использованием вебхуков dbt Cloud и Zapier, аналогично встроенной интеграции со Slack.

Когда задание dbt Cloud завершает выполнение, интеграция будет:

• Получать уведомление вебхука в Zapier,
• Извлекать результаты из API администратора dbt Cloud, и
• Отправлять сводку в канал Microsoft Teams.

Предварительные требования

Для настройки интеграции вам потребуется знание:

• вебхуков dbt Cloud
• Zapier

Настройка соединения между Zapier и Microsoft Teams

• Установите приложение Zapier в Microsoft Teams и предоставьте Zapier доступ к вашему аккаунту.

Примечание: Чтобы получать сообщения, добавьте приложение Zapier в канал команды во время установки.

Создание нового Zap в Zapier

Используйте Webhooks by Zapier в качестве триггера и Catch Raw Hook в качестве события. Если вы не собираетесь проверять подлинность вашего вебхука (не рекомендуется!), то можете выбрать Catch Hook вместо этого.

Нажмите Continue, затем скопируйте URL вебхука.

3. Настройка нового вебхука в dbt Cloud

См. Создание подписки на вебхук для получения полных инструкций. Выберите либо Run completed, либо Run errored, но не оба, иначе вы получите двойные сообщения, когда выполнение завершится с ошибкой.

Запомните секретный ключ вебхука для дальнейшего использования.

После тестирования конечной точки в dbt Cloud вернитесь в Zapier и нажмите Test Trigger, что создаст пример тела вебхука на основе тестового события, отправленного dbt Cloud.

Значения в примере тела жестко закодированы и не отражают ваш проект, но они дают Zapier правильно сформированный объект во время разработки.

Хранение секретов

На следующем шаге вам понадобится секретный ключ вебхука из предыдущего шага и персональный токен доступа или токен сервисного аккаунта dbt Cloud.

Zapier позволяет хранить секреты, что предотвращает отображение ваших ключей в открытом виде в коде Zap. Вы сможете получить к ним доступ через утилиту StoreClient.

Это руководство предполагает, что имена секретных ключей: DBT_CLOUD_SERVICE_TOKEN и DBT_WEBHOOK_KEY. Если вы используете другие имена, убедитесь, что обновили все ссылки на них в примерах кода.

В этом руководстве используется краткосрочное действие для хранения секретов, но вы также можете использовать такие инструменты, как Postman, для взаимодействия с REST API или создать отдельный Zap и вызвать Set Value Action.

a. Создайте подключение Storage by Zapier

Если у вас его еще нет, перейдите на https://zapier.com/app/connections/storage и создайте новое подключение. Запомните сгенерированный секретный UUID для дальнейшего использования.

b. Добавьте временный шаг с кодом

Выберите Run Python в качестве события. Запустите следующий код:

store = StoreClient('abc123') #замените на ваш секретный UUID
store.set('DBT_WEBHOOK_KEY', 'abc123') #замените на секрет вебхука
store.set('DBT_CLOUD_SERVICE_TOKEN', 'abc123') #замените на ваш токен API dbt Cloud

Протестируйте шаг. Вы можете удалить это действие, когда тест пройдет успешно. Ключ будет оставаться сохраненным, если к нему обращаются хотя бы раз в три месяца.

Добавление действия кода

Выберите Code by Zapier в качестве приложения и Run Python в качестве события.

В области Set up action добавьте два элемента в Input Data: raw_body и auth_header. Свяжите их с полями 1. Raw Body и 1. Headers Http Authorization из шага Catch Raw Hook выше.

В поле Code вставьте следующий код, заменив YOUR_SECRET_HERE на секрет, который вы создали при настройке интеграции Storage by Zapier. Помните, что это не ваш секрет dbt Cloud.

Код ниже проверит подлинность запроса, извлечет журналы выполнения для завершенного задания из Admin API, а затем создаст сводное сообщение, которое выделит любые сообщения об ошибках из журналов, созданных dbt Core в конце выполнения.

import hashlib
import hmac
import json
import re


auth_header = input_data['auth_header']
raw_body = input_data['raw_body']

# Доступ к секретным учетным данным
secret_store = StoreClient('YOUR_SECRET_HERE')
hook_secret = secret_store.get('DBT_WEBHOOK_KEY')
api_token = secret_store.get('DBT_CLOUD_SERVICE_TOKEN')

# Проверка, что вебхук пришел из dbt Cloud
signature = hmac.new(hook_secret.encode('utf-8'), raw_body.encode('utf-8'), hashlib.sha256).hexdigest()

if signature != auth_header:
  raise Exception("Calculated signature doesn't match contents of the Authorization header. This webhook may not have been sent from dbt Cloud.")

full_body = json.loads(raw_body)
hook_data = full_body['data'] 

# Шаги, полученные из этих команд, не будут иметь своих деталей ошибок, показанных в строке, так как они неаккуратны
commands_to_skip_logs = ['dbt source', 'dbt docs']

# При тестировании вы захотите жестко закодировать run_id и account_id в существующие ID; пример вебхука не сработает. 
run_id = hook_data['runId']
account_id = full_body['accountId']

# Получение информации о выполнении из Admin API dbt Cloud
url = f'https://YOUR_ACCESS_URL/api/v2/accounts/{account_id}/runs/{run_id}/?include_related=["run_steps"]'
headers = {'Authorization': f'Token {api_token}'}
run_data_response = requests.get(url, headers=headers)
run_data_response.raise_for_status()
run_data_results = run_data_response.json()['data']

# Общая сводка выполнения
outcome_message = f"""
**[{hook_data['runStatus']} for Run #{run_id} on Job \"{hook_data['jobName']}\"]({run_data_results['href']})**


**Environment:** {hook_data['environmentName']} | **Trigger:** {hook_data['runReason']} | **Duration:** {run_data_results['duration_humanized']}

"""

# Сводки по конкретным шагам
for step in run_data_results['run_steps']:
  if step['status_humanized'] == 'Success':
    outcome_message += f"""
✅ {step['name']} ({step['status_humanized']} in {step['duration_humanized']})
"""
  else:
    outcome_message += f"""
❌ {step['name']} ({step['status_humanized']} in {step['duration_humanized']})
"""
    show_logs = not any(cmd in step['name'] for cmd in commands_to_skip_logs)
    if show_logs:
      full_log = step['logs']
      # Удаление временной метки и любых цветовых тегов
      full_log = re.sub('\x1b?\[[0-9]+m[0-9:]*', '', full_log)
    
      summary_start = re.search('(?:Completed with \d+ error.* and \d+ warnings?:|Database Error|Compilation Error|Runtime Error)', full_log)
    
      line_items = re.findall('(^.*(?:Failure|Error) in .*\n.*\n.*)', full_log, re.MULTILINE)
    
      if len(line_items) == 0:
        relevant_log = f'```{full_log[summary_start.start() if summary_start else 0:]}```'
      else:
        relevant_log = summary_start[0]
        for item in line_items:
          relevant_log += f'\n```\n{item.strip()}\n```\n'
      outcome_message += f"""
{relevant_log}
"""

# Zapier ищет словарь `output` для использования в последующих шагах
output = {'outcome_message': outcome_message}

Добавление действия Microsoft Teams

Выберите Microsoft Teams в качестве приложения и Send Channel Message в качестве действия.

В области Set up action выберите команду и канал. Установите Message Text Format в markdown, затем поместите 2. Outcome Message из Run Python в Code by Zapier в поле Message Text.

Тестирование и развертывание

Поскольку вы прошли каждый шаг, вы должны были протестировать выводы, так что теперь вы можете попробовать отправить сообщение в ваш канал Teams.

Когда вы будете довольны результатом, не забудьте убедиться, что ваши run_id и account_id больше не жестко закодированы, затем опубликуйте ваш Zap.

Другие заметки

• Если вы отправляете сообщение в чат, а не в канал команды, вам не нужно добавлять приложение Zapier в Microsoft Teams.
• Если вы отправляете сообщение в чат, а не в канал команды, обратите внимание, что markdown не поддерживается, и вам нужно будет удалить форматирование markdown.
• Если вы выбрали триггер Catch Hook вместо Catch Raw Hook, вам нужно будет передать каждое необходимое свойство из вебхука в качестве входных данных вместо выполнения json.loads() для необработанного тела. Вам также нужно будет удалить код проверки.