SQLMesh 自定义通知渠道：把告警接到钉钉，以及任意 Webhook API

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */
#content_views .toc,
/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */
#content_views.markdown_views > p:empty:has(+ .toc),
#content_views.markdown_views > .toc + p:empty,
/* 富文本旧版目录标记 */
#content_views.htmledit_views #main-toc,
#content_views.htmledit_views #hr-toc,
#content_views.htmledit_views p[id*=\”-toc\”] {
display: none !important;
}
/* 目录去掉后，紧跟的首个标题不再多出一块上边距 */
#content_views.markdown_views > .toc + h1,
#content_views.markdown_views > .toc + h2,
#content_views.markdown_views > .toc + h3,
#content_views.markdown_views > .toc + h4,
#content_views.markdown_views > .toc + p:empty + h1,
#content_views.markdown_views > .toc + p:empty + h2,
#content_views.markdown_views > .toc + p:empty + h3,
#content_views.markdown_views > .toc + p:empty + h4 {
margin-top: 0 !important;
}

上一篇《只用 YAML 配置和 SQL 模型搭告警》里留了个尾巴：SQLMesh 内置的 IM 通道只有 Slack，国内团队基本用不上；内置 SMTP 又只支持 SSL 465。

这一篇专讲自定义通知渠道：钉钉、企业微信、飞书，以及任何「配个 URL 就能 POST JSON」的 Webhook API（自建告警平台、钉钉/飞书的连接器）。

本文关于 SQLMesh 内部结构的所有结论，来自逐行阅读源码 notification_target.py；三家 IM 的接口格式来自各自开放平台官方文档。文末标注了未实测的部分。

一、先看清有哪些路，再动手

读完源码，可用的路线一共三条，各自的代价差得很远：

1.1 为什么不能直接用 slack_webhook 打钉钉

这是最容易产生的误解，得说清楚。源码里 SlackWebhookNotificationTarget 的发送逻辑是：

def _send_slack_message(self, composed: slack.TSlackMessage) –> None:
self.client.send(
text=composed["text"],
blocks=composed["blocks"],
attachments=composed["attachments"],
)

self.client 是 slack_sdk.WebhookClient，它 POST 出去的 JSON 是 Slack 专有格式：

{
"text": "SQLMesh run failed.",
"blocks": [ {"type": "header", …}, {"type": "divider"}, … ],
"attachments": [ … ]
}

而钉钉要的是：

{"msgtype": "text", "text": {"content": "…"}}

字段名、层级、必填项完全对不上。钉钉收到 {"text": …, "blocks": …} 会直接报参数错误（缺少 msgtype）。飞书更不行，它的顶层字段叫 msg_type。

结论：slack_webhook 不是通用 Webhook 客户端，它是个 Slack 客户端。 想让 SQLMesh 直连钉钉，必须自己构造 payload——也就是路线 A 或 B。

不过它仍然有价值：如果钉钉侧前面挂一个「能接收 Slack 格式并转发」的网关，就能零改动复用。这正是路线 A。

1.2 陷阱：type: generic 什么都不做

SQLMesh 源码里有个 GenericNotificationTarget 类，名字看着就像「通用 Webhook 目标」，很容易让人以为在 YAML 里写 type: generic + url: xxx 就能用。不行。

它的类注释原文是：

class GenericNotificationTarget(BaseNotificationTarget):
"""A generic notification target that can be used to create custom notification targets.

This target is not meant to be used directly, but rather as a base class for
custom notification targets. The `send` method should be overridden to provide
the actual notification functionality.
"""

翻译：这个类不用于直接使用，而是作为自定义通知目标的基类；应当重写 send 方法来提供实际的通知功能。

而且它没有定义 url 字段，也没有重写 send()。send() 继承自 BaseNotificationTarget，而那个方法体里只有 docstring——是个空实现：

class BaseNotificationTarget(PydanticModel, frozen=True):
def send(self, notification_status: NotificationStatus, msg: str, **kwargs: t.Any) –> None:
"""Sends notification with the provided message.
…
""" # ← 没有函数体，等于 no-op

所以在 YAML 里写：

notification_targets:
– type: generic
notify_on: [run_failure]
url: https://oapi.dingtalk.com/robot/send?access_token=xxx # ← 会被 pydantic 拒绝或忽略

结果是配置不报错、run 也正常跑、但一条通知都收不到。这类静默失效最难排查，先记住。

GenericNotificationTarget 的正确用法是继承它，这正是路线 B。

1.3 另一个硬限制：YAML 无法引用自定义类

通知目标的类型在源码里是写死的可辨识联合：

NotificationTarget = t.Annotated[
t.Union[
BasicSMTPNotificationTarget,
GenericNotificationTarget,
ConsoleNotificationTarget,
SlackApiNotificationTarget,
SlackWebhookNotificationTarget,
],
Field(discriminator="type_"),
]

只有这 5 个成员，靠 type_ 字段区分，没有插件注册机制。所以：

❌ YAML 里的 type: 只能匹配上述 5 个字面量，无法指向你自己写的类。 ✅ 自定义通知目标必须写在 config.py 里，直接传 Python 实例。

这意味着一旦走路线 B，你的项目就得有 config.py（可以和 config.yml 共存，但通知部分要迁到 Python）。这是路线 A 相对 B 的最大优势——A 能保住纯 YAML 配置。

二、路线 B：一个配置驱动的通用 Webhook 目标

先看主菜。这个类的设计目标是：任何「POST JSON 到某个 URL」的告警接口，都能靠改配置接上，不用再写代码。

平台差异全部收敛到三个点：请求体模板、加签方式、成功判定字段。

2.1 完整实现

保存为项目下的 webhook_target.py：

"""SQLMesh 自定义通知目标：配置驱动的通用 Webhook 渠道。"""
from __future__ import annotations

import base64
import hashlib
import hmac
import json
import logging
import platform as _platform
import socket
import time
import typing as t
import urllib.parse

import requests
from pydantic import Field, SecretStr

from sqlmesh import __version__ as SQLMESH_VERSION
from sqlmesh.core.notification_target import (
GenericNotificationTarget,
NotificationStatus,
)
from sqlmesh.utils.errors import AuditError

logger = logging.getLogger(__name__)

_STATUS_ICON = {
NotificationStatus.SUCCESS: "✅",
NotificationStatus.FAILURE: "🔴",
NotificationStatus.WARNING: "⚠️",
NotificationStatus.INFO: "ℹ️",
NotificationStatus.PROGRESS: "⏳",
}

# 各平台消息体上限（留出余量：异常栈很容易撑爆）
_BYTE_LIMITS = {"dingtalk": 1800, "wecom": 1800, "feishu": 18000}

class WebhookNotificationTarget(GenericNotificationTarget):
"""把 SQLMesh 通知 POST 到任意 Webhook API。

继承 GenericNotificationTarget 并复用其 type_ = "generic"，
因此可作为 notification_targets 列表成员在 config.py 中直接使用。
"""

# —- 必填 —-
url: t.Optional[str] = None

# —- 平台预设：dingtalk / wecom / feishu / raw —-
# raw = 完全由 body_template 决定，用于自建告警平台等任意接口
platform: t.Literal["dingtalk", "wecom", "feishu", "raw"] = "raw"

# —- 消息内容 —-
project: str = "SQLMesh" # 消息前缀，多项目共用一个群时便于区分
title: str = "SQLMesh 告警" # 支持标题的平台（飞书卡片/钉钉 markdown）会用
at_all: bool = False
at_mobiles: t.Tuple[str, ...] = () # 钉钉/企微：按手机号 @
mentioned_list: t.Tuple[str, ...] = () # 企微：按 userid @

# —- 请求控制 —-
method: str = "POST"
headers: t.Dict[str, str] = Field(default_factory=dict)
timeout: int = 10
retries: int = 2 # 额外重试次数

# —- platform="raw" 时的请求体模板（JSON 字符串，支持占位符）—-
body_template: str = ""
# {severity} 占位符的取值来源：把通知状态映射成告警平台的严重级别。
# 默认按 PagerDuty/OpsGenie 常用词表；接自建平台时按需改写。
severity_map: t.Dict[str, str] = Field(default_factory=lambda: {
"failure": "critical",
"warning": "warning",
"success": "info",
"info": "info",
"progress": "info",
})

# —- 加签 —-
secret: t.Optional[SecretStr] = None
# raw 模式下也可指定签名风格；留空则按 platform 自动推断
sign_style: t.Literal["", "dingtalk", "feishu"] = ""

# —- 成功判定 —-
# 三家平台业务失败时都返回 HTTP 200，错误码只在 body 里，必须校验！
success_field: str = "errcode" # 飞书用 "code"
success_value: t.Any = 0

# —- 行为开关 —-
raise_on_error: bool = False # 生产强烈建议 False
truncate_body: bool = True

# ========================================================== 入口

def send(
self,
notification_status: NotificationStatus,
msg: str,
audit_error: t.Optional[AuditError] = None,
exc: t.Optional[str] = None,
**kwargs: t.Any,
) –> None:
if not self.url:
logger.warning("WebhookNotificationTarget 未配置 url，跳过通知")
return

# 1) 组装正文
# audit_error 是 AuditError 对象，可取结构化字段，比 str() 信息更全
if audit_error is not None:
detail = (
f"审计名称: {audit_error.audit_name}\\n"
f"模型: {audit_error.model_name}\\n"
f"违规行数: {audit_error.count}\\n"
f"审计 SQL:\\n{audit_error.sql(pretty=True)}"
)
elif exc:
detail = exc
else:
detail = None

icon = _STATUS_ICON.get(notification_status, "")
text = f"{icon} [{self.project}] {msg}"
if detail:
text = f"{text}\\n\\n{detail}"
if self.truncate_body:
text = self._truncate(text)

status_value = (
notification_status.value
if hasattr(notification_status, "value")
else str(notification_status)
)

# 2) 发送：告警通道自身故障绝不能拖垮主流程
last_err: t.Optional[Exception] = None
for attempt in range(self.retries + 1):
try:
self._post(text, status_value)
return
except Exception as e: # noqa: BLE001
last_err = e
logger.warning(
"SQLMesh 通知发送失败（第 %d/%d 次）: %s",
attempt + 1, self.retries + 1, e,
)
if attempt < self.retries:
time.sleep(min(2 ** attempt, 5)) # 指数退避，最多等 5 秒

logger.error("SQLMesh 通知最终发送失败: %s", last_err)
if self.raise_on_error:
raise last_err # type: ignore[misc]

# ========================================================== HTTP

def _post(self, text: str, status_value: str) –> None:
endpoint = t.cast(str, self.url)
payload: dict[str, t.Any] = {}
secret = self.secret.get_secret_value() if self.secret else ""

# — 加签：钉钉拼 URL query，飞书放 body —
style = self.sign_style or (
"dingtalk" if self.platform == "dingtalk"
else "feishu" if self.platform == "feishu"
else ""
)
if secret and style == "dingtalk":
ts, sign = self._dingtalk_sign(secret)
sep = "&" if "?" in endpoint else "?"
endpoint = f"{endpoint}{sep}timestamp={ts}&sign={sign}"
elif secret and style == "feishu":
ts, sign = self._feishu_sign(secret)
payload.update({"timestamp": ts, "sign": sign})

# — 请求体 —
if self.platform == "raw":
payload.update(self._render_raw(text, status_value))
else:
payload.update(self._im_payload(text))

headers = {"Content-Type": "application/json", **self.headers}
resp = requests.request(
self.method.upper(),
endpoint,
data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
headers=headers,
timeout=self.timeout,
)
resp.raise_for_status()

# — 必须检查 body 里的业务码：三家签名错误都返回 HTTP 200 —
try:
body = resp.json()
except ValueError:
raise RuntimeError(f"响应非 JSON: {resp.text[:200]}")

if isinstance(body, dict) and self.success_field in body:
if body[self.success_field] != self.success_value:
raise RuntimeError(f"Webhook 返回业务错误: {body}")

# ————————————————- 国内 IM 请求体

def _im_payload(self, text: str) –> dict[str, t.Any]:
if self.platform == "dingtalk":
# 用 text 而非 markdown：markdown 的 @ 需在正文额外写一遍手机号才生效，
# 只填 at.atMobiles 不会真正 @ 到人，极易配错。
return {
"msgtype": "text",
"text": {"content": text},
"at": {
"atMobiles": list(self.at_mobiles),
"isAtAll": self.at_all,
},
}

if self.platform == "wecom":
return {
"msgtype": "text",
"text": {
"content": text,
"mentioned_list": list(self.mentioned_list),
"mentioned_mobile_list": list(self.at_mobiles),
},
}

# 飞书：@ 人只能内嵌在正文里
if self.at_all:
text = f'<at user_id="all">所有人</at> {text}'
return {
"msg_type": "text", # 注意：字段名与前两家不同
"content": {"text": text}, # 注意：层级与前两家不同
}

# ————————————————- 任意 Webhook 模板

def _render_raw(self, text: str, status: str) –> dict[str, t.Any]:
"""用 body_template 渲染任意 Webhook API 的请求体。

模板是一段 JSON 字符串，支持占位符：
{status} 通知状态：success / failure / warning / info / progress
{severity} 由 status 映射出的严重级别，可直接用于告警平台分级
{msg} {project} {title} {timestamp}
{host} {sqlmesh_version} {python_version}

安全说明：只做字符串替换，不使用 eval/format 执行任意表达式；
替换值经过 JSON 转义，消息里的换行和引号不会破坏结构。
"""
template = self.body_template
if not template:
# 未提供模板时退化为最通用的形状：多数告警网关都接受裸 text 字段
return {"text": text, "status": status}

mapping = {
"status": status,
"severity": self.severity_map.get(status, "info"),
"msg": text,
"project": self.project,
"title": self.title,
"timestamp": str(int(time.time())),
"host": socket.gethostname(),
"sqlmesh_version": SQLMESH_VERSION,
"python_version": _platform.python_version(),
}
rendered = template
for key, value in mapping.items():
# json.dumps(…)[1:-1] 得到已转义的字符串内容，可安全嵌入 JSON
rendered = rendered.replace(
"{" + key + "}", json.dumps(str(value), ensure_ascii=False)[1:–1]
)
return t.cast(dict[str, t.Any], json.loads(rendered))

# ========================================================== 加签

@staticmethod
def _dingtalk_sign(secret: str) –> tuple[str, str]:
"""钉钉：毫秒时间戳；key=secret，msg=timestamp\\\\nsecret；结果需 urlencode。"""
timestamp = str(round(time.time() * 1000))
string_to_sign = f"{timestamp}\\n{secret}"
digest = hmac.new(
secret.encode("utf-8"),
string_to_sign.encode("utf-8"),
digestmod=hashlib.sha256,
).digest()
return timestamp, urllib.parse.quote_plus(base64.b64encode(digest))

@staticmethod
def _feishu_sign(secret: str) –> tuple[str, str]:
"""飞书：秒级时间戳；key=timestamp\\\\nsecret，msg=空字符串；结果不 urlencode。

与钉钉正好把 HMAC 的 key/msg 角色对调，官方 Java/Go 示例同为对空数据签名。
"""
timestamp = str(int(time.time()))
string_to_sign = f"{timestamp}\\n{secret}"
digest = hmac.new(
string_to_sign.encode("utf-8"),
digestmod=hashlib.sha256,
).digest()
return timestamp, base64.b64encode(digest).decode("utf-8")

# ========================================================== 工具

def _truncate(self, text: str) –> str:
"""按平台字节上限截断，保留头尾。

Python 异常栈的特点是：开头是结论，结尾最内层 traceback 才是根因，
中间几十层框架调用栈信息量最低。所以保头保尾、砍中间。
"""
limit = _BYTE_LIMITS.get(self.platform, 2000)
encoded = text.encode("utf-8")
if len(encoded) <= limit:
return text
head = limit * 2 // 3
tail = max(limit – head – 60, 0)
return (
encoded[:head].decode("utf-8", errors="ignore")
+ "\\n\\n…[内容过长已截断]…\\n\\n"
+ encoded[–tail:].decode("utf-8", errors="ignore")
)

@property
def is_configured(self) –> bool:
return bool(self.url)

📌 三个内置平台（钉钉/企微/飞书）直接用 platform + _im_payload()，不需要碰 body_template；只有接任意第三方 JSON 接口时才用 platform="raw" + body_template，见 2.3。

2.2 挂到 config.py

import os
from pydantic import SecretStr

from sqlmesh.core.config import Config, ModelDefaultsConfig
from sqlmesh.core.notification_target import BasicSMTPNotificationTarget
from webhook_target import WebhookNotificationTarget

config = Config(
model_defaults=ModelDefaultsConfig(dialect="duckdb"),
notification_targets=[
# 主力：钉钉群，失败类事件全推 + @ 值班
WebhookNotificationTarget(
notify_on=[
"run_failure",
"apply_failure",
"audit_failure",
"migration_failure",
],
platform="dingtalk",
url=os.getenv("DINGTALK_WEBHOOK"),
secret=SecretStr(os.getenv("DINGTALK_SECRET", "")),
project="数据仓库-sushi",
at_mobiles=("13800000000",),
success_field="errcode",
success_value=0,
raise_on_error=False, # 告警失败不能拖垮管道
),

# 备份：邮件留档（IM 消息会被刷走，邮件可检索）
BasicSMTPNotificationTarget(
notify_on=["run_failure", "apply_failure"],
host=os.getenv("SMTP_HOST"),
port=465, # 内置实现固定 SMTP_SSL
user=os.getenv("SMTP_USER"),
password=SecretStr(os.getenv("SMTP_PASSWORD", "")),
sender="data-alert@example.com",
recipients=["data-team@example.com"],
subject="[数据仓库] SQLMesh 管道告警",
),
],
)

2.3 接自建告警平台 / 任意 Webhook API

这就是「支持配置 webhook API」的核心用法。假设你的告警网关要求这样的请求体：

{
"source": "sqlmesh",
"severity": "critical",
"service": "数据仓库-sushi",
"summary": "SQLMesh run failed.",
"detail": "Traceback …",
"host": "prod-etl-01"
}

定义一个子类，把模板写死即可：

import typing as t

from webhook_target import WebhookNotificationTarget

class OpsGenieLikeTarget(WebhookNotificationTarget):
"""对接自建告警平台 Webhook / 任意 JSON 接口。"""

platform: t.Literal["raw"] = "raw"
success_field: str = "status" # 按你的接口返回改
success_value: t.Any = "ok"

# {severity} 会按通知状态自动映射（failure→critical、success→info…），
# 不必为不同事件写多个子类。映射表见基类 severity_map，可覆盖。
body_template: str = """
{
"source": "sqlmesh",
"severity": "{severity}",
"status": "{status}",
"service": "{project}",
"summary": "{title}",
"detail": "{msg}",
"host": "{host}",
"ts": "{timestamp}",
"sqlmesh_version": "{sqlmesh_version}"
}
"""

📌 {severity} 的映射由基类 severity_map 决定，默认 failure→critical、warning→warning、其余 →info。接自建平台时若用别的词表（比如 P1/P2），直接覆盖这个字段：

severity_map: t.Dict[str, str] = {"failure": "P1", "warning": "P2",
"success": "P4", "info": "P4", "progress": "P4"}

配置：

OpsGenieLikeTarget(
notify_on=["run_failure", "apply_failure", "audit_failure"],
url="https://alert.internal.example.com/api/v1/events",
headers={"Authorization": f"Bearer {os.getenv('ALERT_TOKEN')}"},
project="数据仓库-sushi",
title="SQLMesh 管道失败",
timeout=15,
retries=3,
)

可用占位符：{status} {msg} {project} {title} {timestamp} {host} {sqlmesh_version} {python_version}。

🔐 关于模板实现的安全性：这里刻意只用字符串替换，不用 str.format() 或 eval。原因是 {msg} 里塞的是 Python 异常栈，内容完全不可控（可能包含大括号、引号、换行）。用 str.format() 会因为异常栈里的 {} 抛 KeyError/IndexError，甚至被构造成属性访问；eval 更是直接的代码注入。所有替换值都先经 json.dumps(…)[1:-1] 转义，保证换行和引号不会破坏 JSON 结构。如果你自己改写这段，请务必保留这个转义步骤。

三、三家国内 IM 的接口差异对照

三家都是「群机器人 Webhook + POST JSON」，但加签算法、时间戳单位、签名放哪、消息字段名全都不一样。这是真正的坑所在。

3.1 钉钉加签

官方算法：把 timestamp + "\\n" + 密钥 当签名字符串，HmacSHA256 计算，Base64 编码，再 urlEncode，最后拼到 URL：

https://oapi.dingtalk.com/robot/send?access_token=xxx&timestamp=1730000000000&sign=xxxxx

🔥 两个高频翻车点（社区踩坑记录里最常见）：

3.2 企业微信（最简单）

不需要任何签名，key 已经在 Webhook URL 里，直接 POST：

{
"msgtype": "text",
"text": {
"content": "🔴 [数据仓库-sushi] SQLMesh run failed.\\n…",
"mentioned_list": ["zhangsan", "@all"],
"mentioned_mobile_list": ["13800000000"]
}
}

📌 markdown / markdown_v2 类型不支持 mentioned_list 这套 @ 语法。text 和 markdown 支持在 content 里用 <@userid> 扩展语法 @ 群成员（markdown_v2 不支持）。想同时要 @ 人和排版，用 markdown + <@userid>。

3.3 飞书加签（参数角色与钉钉相反）

飞书官方的 Python 示例长这样，第一次看会以为是文档印错了：

def gen_sign(timestamp, secret):
string_to_sign = '{}\\n{}'.format(timestamp, secret)
# 注意：hmac.new 只传了 key，没传 msg！
hmac_code = hmac.new(string_to_sign.encode("utf-8"), digestmod=hashlib.sha256).digest()
sign = base64.b64encode(hmac_code).decode('utf-8')
return sign

它确实是「用 timestamp\\nsecret 作为 key，对空字符串做 HMAC-SHA256」——和钉钉正好把 key/msg 的角色对调。这不是笔误：官方 Java 示例是 mac.doFinal(new byte[]{})，Go 示例是 h.Write(data) 且 data 为空切片，都是对空数据签名。

请求体（签名在 body，不在 URL）：

{
"timestamp": "1730000000",
"sign": "xxxxxxxxxxxxxxxx",
"msg_type": "text",
"content": {"text": "…"}
}

飞书官方列出的签名失败原因（code: 19021）值得记住：

• 时间戳距发送时刻超过 1 小时（签名过期）
• 服务器时间与标准时间偏差过大 —— 容器/虚机没做 NTP 同步时特别容易中招
• 签名算错（多半是把 key/msg 按钉钉习惯写反了）

⚠️ 飞书的一个额外成本：飞书自定义机器人没有任何数据访问权限，所以它自己无法通过邮箱/手机号反查 open_id。而 @ 人要求 user_id 必须是有效的 Open ID 或 User ID。官方给的两条获取路径都需要额外开发一个自建应用（申请 contact:user.id:readonly 反查，或订阅消息事件解析）。

也就是说：「失败时 @ 到具体责任人」这个需求，飞书的实现成本明显高于钉钉和企微——钉钉直接填手机号，企微直接填 userid。

四、路线 A：零改动 SQLMesh，用适配器服务转发

如果你不想让 SQLMesh 项目里出现 Python（保持纯 config.yml），或者想让多个数据平台共用一个告警网关，这条路线更合适。

4.1 原理

利用 slack_webhook 只是个「往 URL POST JSON 的客户端」这一事实，把它的 url 指向你自己的服务：

SQLMesh ──(Slack 格式 JSON)──▶ 你的适配器服务 ──(钉钉格式 JSON)──▶ 钉钉群

SQLMesh 侧配置（纯 YAML，完全不动）：

notification_targets:
– type: slack_webhook
notify_on:
– run_failure
– apply_failure
– audit_failure
url: "https://alert-adapter.internal.example.com/slack-to-dingtalk"

注意：这条路需要装 Slack 依赖，因为 WebhookClient 来自 slack_sdk。源码里缺依赖会抛 MissingDependencyError: Missing Slack dependencies. Run pip install 'sqlmesh[slack]' to install them.

pip install 'sqlmesh[slack]'

装它只是为了拿到那个 HTTP 客户端，并不代表真的在用 Slack。

4.2 适配器服务（示例，FastAPI）

"""把 Slack Webhook 格式的消息转成钉钉格式。独立部署，与 SQLMesh 项目无关。"""
import base64
import hashlib
import hmac
import os
import time
import urllib.parse

import requests
from fastapi import FastAPI, Request

app = FastAPI()

DINGTALK_WEBHOOK = os.environ["DINGTALK_WEBHOOK"]
DINGTALK_SECRET = os.environ.get("DINGTALK_SECRET", "")
AT_MOBILES = [m for m in os.environ.get("AT_MOBILES", "").split(",") if m]

def dingtalk_sign(secret: str) –> tuple[str, str]:
timestamp = str(round(time.time() * 1000)) # 毫秒
string_to_sign = f"{timestamp}\\n{secret}"
digest = hmac.new(
secret.encode("utf-8"),
string_to_sign.encode("utf-8"),
digestmod=hashlib.sha256,
).digest()
return timestamp, urllib.parse.quote_plus(base64.b64encode(digest))

@app.post("/slack-to-dingtalk")
async def slack_to_dingtalk(request: Request):
payload = await request.json()

# slack_sdk 发来的结构：{"text": …, "blocks": […], "attachments": […]}
# text 字段就是纯文本消息，正好够用
text = payload.get("text") or ""
if not text:
# 兜底：从 blocks 里捞文本
for block in payload.get("blocks") or []:
if block.get("type") == "section":
text += (block.get("text") or {}).get("text", "") + "\\n"

endpoint = DINGTALK_WEBHOOK
if DINGTALK_SECRET:
ts, sign = dingtalk_sign(DINGTALK_SECRET)
endpoint = f"{DINGTALK_WEBHOOK}&timestamp={ts}&sign={sign}"

resp = requests.post(
endpoint,
json={
"msgtype": "text",
"text": {"content": text[:1900]}, # 钉钉 text 上限 2048 字节
"at": {"atMobiles": AT_MOBILES, "isAtAll": False},
},
headers={"Content-Type": "application/json"},
timeout=10,
)
return {"status": resp.status_code, "body": resp.json()}

4.3 两条路线怎么选

经验判断：单个项目、想尽快上线 → 路线 B。多个数据平台、已有告警网关、或生产环境不能直连公网 → 路线 A。

路线 A 还有一个隐藏优势：SQLMesh 只需能访问内网地址。很多企业生产环境不允许直连 oapi.dingtalk.com，但允许访问内网服务，适配器正好架在能出公网的 DMZ 机器上。

五、顺带解决：企业邮箱只有 587/STARTTLS

上一篇提过内置 smtp 目标硬编码 smtplib.SMTP_SSL。源码确认：

class BasicSMTPNotificationTarget(BaseTextBasedNotificationTarget):
host: t.Optional[str] = None
port: int = 465 # 默认 465
user: t.Optional[str] = None
password: t.Optional[SecretStr] = None
sender: t.Optional[str] = None
recipients: t.Optional[t.FrozenSet[str]] = None
subject: t.Optional[str] = "SQLMesh Notification" # 文档页未提，源码支持
type_: t.Literal["smtp"] = Field(alias="type", default="smtp")

def send_text_message(self, notification_status, msg) –> None:
...
with smtplib.SMTP_SSL(host=self.host, port=self.port) as smtp: # 强制 SSL

要点：只支持隐式 SSL（465），不支持 STARTTLS（587）和明文（25）。port 可改，但协议固定。

如果你的邮件服务器只有 587，继承重写即可。这里用 BaseTextBasedNotificationTarget，它已经帮你把 msg 和 exc / str(audit_error) 拼好了，只需实现一个方法：

import smtplib
from email.message import EmailMessage

from sqlmesh.core.notification_target import (
BasicSMTPNotificationTarget,
NotificationStatus,
)
from sqlmesh.utils.errors import ConfigError

class StartTLSNotificationTarget(BasicSMTPNotificationTarget):
"""支持 STARTTLS(587) 与明文(25) 的 SMTP 通知目标。"""

use_tls: bool = True # 587 → True；25 且服务器允许明文 → False

def send_text_message(
self, notification_status: NotificationStatus, msg: str
) –> None:
if not self.host:
raise ConfigError("Missing SMTP host for notification")

email = EmailMessage()
email["Subject"] = f"{self.subject} [{notification_status.value}]"
email["To"] = ",".join(self.recipients or [])
email["From"] = self.sender
email.set_content(msg)

with smtplib.SMTP(host=self.host, port=self.port, timeout=30) as smtp:
if self.use_tls:
smtp.starttls() # 明文连接升级为 TLS
if self.user and self.password:
smtp.login(self.user, self.password.get_secret_value())
smtp.send_message(email)

配置时 port=587、use_tls=True。

这个例子同时说明：SQLMesh 的内置目标都是普通 Python 类，任何行为都能通过继承改写，不限于「换个 IM 平台」。

六、怎么验证它真的能工作

我没有可运行的 SQLMesh 环境和真实的国内 IM 机器人凭据，本文代码未做端到端实测。 下面是从低成本到高成本的验证路径，建议按顺序走：

第 1 步：用内置 console 目标确认事件被触发

源码里有个 ConsoleNotificationTarget（注释说是留给测试用的），把通知打到控制台，零外部依赖：

notification_targets:
– type: console
notify_on:
– run_start
– run_end
– run_failure
– audit_failure

人为制造一次失败（把某模型 SQL 改成必然报错），跑 sqlmesh run，看控制台有没有输出。这一步能把「我的类有 bug」和「SQLMesh 根本没触发事件」彻底分开。

第 2 步：用 curl 单独验证机器人 Webhook 本身

把「机器人配置问题」和「我的代码问题」再切一刀。企业微信无签名，最适合直接测：

curl -X POST -H "Content-Type: application/json" \\
-d '{"msgtype":"text","text":{"content":"curl 连通性测试"}}' \\
"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxx"

飞书无签名时：

curl -X POST -H "Content-Type: application/json" \\
-d '{"msg_type":"text","content":{"text":"curl 连通性测试"}}' \\
"https://open.feishu.cn/open-apis/bot/v2/hook/xxxx"

成功返回 {"code":0,"msg":"success",…}；格式错返回 {"code":9499,"msg":"Bad Request"}。

第 3 步：脱离 SQLMesh，单独测自定义类

自定义目标是普通 Python 对象，可以直接实例化并手动调用 notify_*：

import os
from pydantic import SecretStr
from webhook_target import WebhookNotificationTarget

target = WebhookNotificationTarget(
notify_on=["run_failure"],
platform="dingtalk",
url=os.getenv("DINGTALK_WEBHOOK"),
secret=SecretStr(os.getenv("DINGTALK_SECRET", "")),
project="连通性测试",
raise_on_error=True, # ← 测试时打开，让错误暴露出来
retries=0,
)

# 等价于 SQLMesh run 失败时的调用
target.notify_run_failure("ConnectionError: warehouse unreachable")

群里收到消息 = 签名算法和 payload 都对了。收不到就看异常里的 HTTP body 业务码。

第 4 步：验证 @ 人、截断、重试

• 用故意超长的异常文本（几十 KB）调用，确认截断生效且平台没拒收。
• 确认 @ 真的触发了（看手机是否收到红点提醒，而不只是正文里有手机号字符串）。
• 把 url 临时改成不通的地址，确认重试和 raise_on_error=False 的静默降级符合预期。

第 5 步：端到端

挂回 config.py，制造真实失败，跑 sqlmesh run，确认群里收到带完整异常栈的告警。

🔐 安全提醒：Webhook 地址和加签密钥等同于「往你群里发消息的权限」。三家平台官方都明确要求不要公布在 GitLab、博客等可公开查阅的地方，泄露后任何人都能往群里刷消息。飞书/钉钉可在机器人设置里直接「重置」密钥。

七、生产环境的六条建议

八、小结

真正的成本不在「能不能自定义」——SQLMesh 的扩展点是干净的、官方在源码注释里就摆着。成本在于三家国内 IM 的接口细节互不兼容：字段名、时间戳单位、HMAC 的 key/msg 角色、签名放 URL 还是 body、三种完全不同的 @ 人语法、三家都返回 200 的业务错误码。把这些差异封装进一个类之后，后续切换平台就只是改一个 platform 字段的事了。

参考来源

SQLMesh

• 通知目标源码 notification_target.py（本文所有内部结构结论的依据）：https://github.com/TobikoData/sqlmesh/blob/main/sqlmesh/core/notification_target.py
• Notifications guide：https://sqlmesh.readthedocs.io/en/stable/guides/notifications/
• 配置参考：https://sqlmesh.readthedocs.io/en/stable/reference/configuration/

钉钉

• 自定义机器人安全设置（加签算法）：https://open.dingtalk.com/document/dingstart/customize-robot-security-settings
• 自定义机器人发送群聊消息：https://open.dingtalk.com/document/development/custom-bot-to-send-group-chat-messages

企业微信

• 消息推送配置说明（消息类型、数据格式、频率限制、<@userid> 语法）：https://developer.work.weixin.qq.com/document/path/91770

飞书

• 自定义机器人使用指南（签名校验、消息类型、@ 人、限流、错误码）：https://open.feishu.cn/document/client-docs/bot-v3/add-custom-bot

关于本文可靠性的说明

SQLMesh 部分的事实（类名、方法签名、字段默认值、事件枚举、类型联合、GenericNotificationTarget 无 url 字段且 send() 为空实现、slack_webhook 发送 text/blocks/attachments 结构、SMTP_SSL 硬编码）均逐条来自我实际读取的 main 分支源码，非推测。不同版本的 SQLMesh 可能有差异，请以你所用版本的源码为准。

三家 IM 的接口格式、限流数值、错误码来自各自官方文档；加签算法按官方文档的 Python 示例原样转录，飞书「对空字符串签名」的行为已用官方 Java/Go 示例交叉验证。