airflow.providers.slack.hooks.slack_webhook¶

屬性¶

LEGACY_INTEGRATION_PARAMS

類¶

SlackWebhookHook

此類提供了對 slack_sdk.WebhookClient 的輕量級封裝。

函式¶

check_webhook_response(func)

檢查 WebhookResponse 並在狀態碼不等於 200 時丟擲錯誤。

模組內容¶

airflow.providers.slack.hooks.slack_webhook.LEGACY_INTEGRATION_PARAMS = ('channel', 'username', 'icon_emoji', 'icon_url')[source]¶

airflow.providers.slack.hooks.slack_webhook.check_webhook_response(func)[source]¶

檢查 WebhookResponse 並在狀態碼不等於 200 時丟擲錯誤。

class airflow.providers.slack.hooks.slack_webhook.SlackWebhookHook(*, slack_webhook_conn_id, timeout=None, proxy=None, retry_handlers=None, **extra_client_args)[source]¶

基類: airflow.hooks.base.BaseHook

此類提供了對 slack_sdk.WebhookClient 的輕量級封裝。

此 Hook 允許您使用 Incoming Webhook 向 Slack 釋出訊息。

另請參閱

注意

當您使用 Incoming Webhook 釋出訊息時，您無法覆蓋預設頻道（由安裝您應用的使用者選擇）、使用者名稱或圖示。相反，這些值將始終繼承自關聯的 Slack 應用配置（連結）。只有在 Legacy Slack Integration Incoming Webhook 中才可能更改這些值。

警告

此 Hook 旨在使用 Slack Incoming Webhook 連線，並且可能無法與 Slack API 連線正常工作。

示例

# Create hook
hook = SlackWebhookHook(slack_webhook_conn_id="slack_default")

# Post message in Slack channel by JSON formatted message
# See: https://api.slack.com/messaging/webhooks#posting_with_webhooks
hook.send_dict({"text": "Hello world!"})

# Post simple message in Slack channel
hook.send_text("Hello world!")

# Use ``slack_sdk.WebhookClient``
hook.client.send(text="Hello world!")

引數:

slack_webhook_conn_id (str) – Slack Incoming Webhook 連線 ID，其密碼欄位中包含 Incoming Webhook token。
timeout (int | None) – 客戶端等待連線並從 Slack 接收響應的最大秒數。如果未設定，則使用預設的 WebhookClient 值。
proxy (str | None) – 用於進行 Slack Incoming Webhook 呼叫的代理。
retry_handlers (list[slack_sdk.http_retry.RetryHandler] | None) – slack_sdk.WebhookClient 中用於自定義重試邏輯的處理程式列表。

conn_name_attr = 'slack_webhook_conn_id'[source]¶

default_conn_name = 'slack_default'[source]¶

conn_type = 'slackwebhook'[source]¶

hook_name = 'Slack Incoming Webhook'[source]¶

slack_webhook_conn_id[source]¶

timeout = None[source]¶

proxy = None[source]¶

retry_handlers = None[source]¶

extra_client_args[source]¶

property client: slack_sdk.WebhookClient[source]¶

獲取底層的 slack_sdk.webhook.WebhookClient (已快取)。

get_conn()[source]¶

獲取底層的 slack_sdk.webhook.WebhookClient (已快取)。

send_dict(body, *, headers=None)[source]¶

使用給定的 JSON 資料塊執行 Slack Incoming Webhook 請求。

引數:

body (dict[str, Any] | str) – JSON 資料結構，預期為 dict 或 JSON 字串。
headers (dict[str, str] | None) – 此請求的請求頭。

send(*, text=None, blocks=None, response_type=None, replace_original=None, delete_original=None, unfurl_links=None, unfurl_media=None, headers=None, attachments=None, **kwargs)[source]¶

使用給定的引數執行 Slack Incoming Webhook 請求。

引數:

text (str | None) – 文字訊息（即使有 blocks，也建議設定此項作為回退）。
blocks (list[dict[str, Any]] | None) – Block Kit UI 元件的集合。
response_type (str | None) – 訊息型別（'in_channel' 或 'ephemeral'）。
replace_original (bool | None) – 如果您對 response_url 請求使用此選項，則為 True。
delete_original (bool | None) – 如果您對 response_url 請求使用此選項，則為 True。
unfurl_links (bool | None) – 指示文字 URL 是否應展開的選項。
unfurl_media (bool | None) – 指示媒體 URL 是否應展開的選項。
headers (dict[str, str] | None) – 此請求的請求頭。
attachments (list[dict[str, Any]] | None) – (舊版) attachments 的集合。

send_text(text, *, unfurl_links=None, unfurl_media=None, headers=None)[source]¶

使用給定的文字執行 Slack Incoming Webhook 請求。

引數:

text (str) – 文字訊息。
unfurl_links (bool | None) – 指示文字 URL 是否應展開的選項。
unfurl_media (bool | None) – 指示媒體 URL 是否應展開的選項。
headers (dict[str, str] | None) – 此請求的請求頭。

classmethod get_connection_form_widgets()[source]¶

返回新增到 Hook 以處理額外值的 widget 字典。

classmethod get_ui_field_behaviour()[source]¶

返回自定義欄位行為。