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]¶

返回自定义字段行为。