[Prefect] SlackWebhook 사용법

Overview

• Prefect Block에서 제공하는 SlackWebhook 사용 방법에 대해 설명합니다.
• 아래의 코드 예제에서는 Flow가 failure 상태로 바뀔 때 Slack을 발송하는 예시이지만 Task에서도 동일하게 사용 가능합니다.
• https://docs.prefect.io/latest/api-ref/prefect/blocks/notifications/?h=slack#prefect.blocks.notifications.SlackWebhook

 

Flow에서 사용 가능한 상태

• on_completion
• on_cancellation
• on_crashed
• on_failure

Task에서 사용 가능한 상태

• on_completion
• on_failure

 

from prefect import variables
from prefect.blocks.notifications import SlackWebhook
from prefect.settings import PREFECT_UI_URL


def flow_failure_webhook(flow, flow_run, state):
    slack_webhook = SlackWebhook(url=variables.get("my_slack_webhook"))
    slack_webhook_block = slack_webhook.load("slack-webhook-block")
    slack_webhook_block.notify(
        (
            f"Flow: {flow}"
            f"Your job {flow_run.name} entered {state.name} "
            f"with message:\n\n"
            f"See <{PREFECT_UI_URL.value()}/flow-runs/"
            f"flow-run/{flow_run.id}|the flow run in the UI>\n\n"
            f"Tags: {flow_run.tags}\n\n"
            f"Scheduled start: {flow_run.expected_start_time}"
        )
    )

# Flow의 상태가 failure 일 때, Slack 발송
@flow(
    name=METRIC_CPU_SCHEDULER_NAME,
    flow_run_name=generate_flow_run_name,
    retries=1,
    retry_delay_seconds=5,
    description="Prefect agent module for CPU usage",
    timeout_seconds=5,
    task_runner=SequentialTaskRunner(),
    on_failure=[flow_failure_webhook],
)