简道云自建插件 + 外部后端服务:一次完整的踩坑实践

一、为什么不让插件直连数据库

简道云的自建插件运行在云端，而我们的业务数据在内网，两者天然不通。即便通，把数据库账号密码写进云端插件代码也很不安全。

所以最稳的架构是：重逻辑放在我们自己托管的后端服务上，简道云插件只当「转发层」——它把表单里的参数通过 HTTP 发给我们的服务，服务查完库再把结果回传，插件负责把结果写回表单字段。

二、踩坑实录（重点）

下面这几条都是实测踩出来的，照着避坑能省很多时间。

坑 1：插件代码里不能用 `data` 当变量名

现象：运行报 name 'data' is not defined。
原因：简道云插件运行环境内部已存在 data 相关变量/保留字，插件代码若把 data 用作普通变量名（如 data = resp.json().get("data")），会与环境冲突导致 NameError。
解决：把承接返回体的局部变量改名，比如 book_info = body.get("data")。只要别叫 data 就行。

坑 2：插件代码是「顶层代码」，不是 `def main()`

错误写法：把逻辑包在 def main(): ... 里——简道云不认这个入口，会报错。
正确写法：插件编辑器里直接写顶层代码，开头 import，中间读入参，结尾用 return {...} 返回字典（key 即出参 ID）。不要写 if __name__ == "__main__"。

坑 3：`globals()` 在插件沙箱里会报语法错误

想兼容不同写法去用 globals().get("triggerConf")？加上就语法错误，沙箱不允许。
平台把入参注入为全局变量 triggerConf，直接引用即可：isbn = triggerConf.get("主键参数")。
调试工具请求体里常写成 {"trigger_conf": {...}}，那只是发包格式，运行时已映射成 triggerConf，插件里按 triggerConf 读就行。

坑 4：SQL 取「最新一条」的写法

给定主键查最新记录，用参数化 + TOP 1 + 排序：

SELECT TOP 1 字段 FROM 表
WHERE 主键 =?
ORDERBY 版本字段 DESC, 序号字段 DESC

用 ? 占位传参（防 SQL 注入），不要再叠加一个 LIKE '前缀%' 的范围过滤，否则会和精确匹配冲突、甚至查不到。
如果「版本/序号」是字符串（如 '10'、'2'），按字符串降序会把 '10' 排到 '2' 前面，排序就错了。稳妥起见用 TRY_CAST(版本字段 AS INT) DESC。

坑 5：后端必须暴露在「公网 HTTPS」

简道云插件服务器要能访问到你的后端，本地 127.0.0.1 / 内网地址不行。
务必部署到公网 + 域名 + HTTPS，并在前面套一层鉴权（固定 Token 或来源 IP 白名单）——这个接口等于把「按主键查库」的能力开放出去，不加鉴权很危险。

坑 6：空值无法直接序列化

查询结果为空的字段，在 Python 里常是 NaN（float 类型），直接返回 JSON 会报错。
返回前把 NaN / None 统一清洗一遍即可。

三、核心代码（已泛化，不含具体业务字段）

3.1 自建后端服务（FastAPI，部署在你自己的服务器）

import pandas as pdimport pyodbcfrom fastapi import FastAPIfrom pydantic import BaseModel, Field# 数据库连接：返回你的 ODBC 连接串（内部实现，此处省略）def connection_string_src() -> str: ...def Sqlquery(query, params=None):    """执行 SQL 返回 DataFrame；VARCHAR 按 GBK 解码，避免中文乱码。"""    conn = pyodbc.connect(connection_string_src())    conn.add_output_converter(        pyodbc.SQL_VARCHAR,        lambda b: b.decode("gbk", errors="replace") if b else None,    )    try:        return pd.read_sql(query, conn, params=params) if params else pd.read_sql(query, conn)    finally:        conn.close()  # 放 finally，异常也不泄漏连接app = FastAPI()# 固定查询：给定业务主键，按版本/序号降序取第一条# （以下 SELECT 列仅作占位示意，实际替换为你的业务字段）BOOK_SQL = """SELECT TOP 1    eta.业务字段1 AS [业务字段1],    eta.业务字段2 AS [业务字段2],    eta.业务字段3 AS [业务字段3]    -- ... 其余业务字段FROM 主表 AS etaLEFT JOIN 关联表1 AS t1 ON eta.键 = t1.键LEFT JOIN 关联表2 AS t2 ON t2.关联键 = t1.关联键WHERE (ISNULL(eta.主键, '') <> '')  AND (eta.主键 = ?)ORDER BY TRY_CAST(eta.版本字段 AS INT) DESC,         TRY_CAST(eta.序号字段 AS INT) DESC"""class KeyRequest(BaseModel):    主键: str = Field(..., description="业务主键，如书号")def _clean(row: dict) -> dict:    """NaN / NaT 转 None，避免 JSON 序列化报错。"""    return {k: (None if (isinstance(v, float) and pd.isna(v)) else v) for k, v in row.items()}@app.post("/api/query")def query(req: KeyRequest):    主键 = (req.主键 or "").strip()    if not 主键:        return {"code": 400, "msg": "主键不能为空", "data": None}    try:        df = Sqlquery(BOOK_SQL, params=[主键])    except Exception as exc:        return {"code": 500, "msg": f"查询失败: {exc}", "data": None}    if df is None or df.empty:        return {"code": 404, "msg": "未找到对应记录", "data": None}    return {"code": 0, "msg": "ok", "data": _clean(df.iloc[0].to_dict())}

3.2 简道云插件「后端函数」代码（整段粘进插件编辑器）

import requestsimport jsonBACKEND_URL = "https://你的公网域名/api/query"# 简道云注入的入参全局变量就是 triggerConf，直接引用（别用 globals()！）try:    _conf = triggerConfexcept NameError:    _conf = {}# 读取主键参数（做好健壮性处理：可能传 字符串/列表/字典）key = _conf.get("主键")try:    if isinstance(key, str) and key.strip().startswith("["):        key = json.loads(key)    if isinstance(key, list):        key = key[0] if key else None    if isinstance(key, dict):        key = key.get("主键") or key.get("value")except Exception:    passif not key or not str(key).strip():    raise ValueError("主键不能为空，请先填写")key = str(key).strip()# 调用自建后端（简道云插件环境自带 requests）try:    resp = requests.post(BACKEND_URL, json={"主键": key}, timeout=10)    body = resp.json()except Exception as e:    raise ValueError("调用后端服务失败: " + str(e))if body.get("code") != 0:    raise ValueError(body.get("msg") or "查询失败")# 注意：这里绝不用 data 作变量名，改用 book_infobook_info = body.get("data") or {}# 返回字典，key 与插件【出参】ID 一一对应return {    "field1": book_info.get("业务字段1"),    "field2": book_info.get("业务字段2"),    "field3": book_info.get("业务字段3"),    # ... 其余字段按相同规律扩展}

四、简道云侧怎么配（通用步骤）

建表单字段：一个「主键」输入字段（绑定插件入参），若干个「结果」字段（绑定插件出参）。
给按钮加前端事件：按钮 → 执行动作 → 添加「前端事件」→ 动作类型选「插件」→ 选你的自建插件。
映射：入参把表单「主键」字段 → 插件入参；出参把插件返回的每个 key → 存到对应表单字段。
发布到企业工作台：表单发布时勾选「企业工作台」可见即可。
用法：工作台打开表单 → 填主键 → 点按钮 → 结果自动回填。

前提：插件在「开放平台 → 开发者后台 → 自建插件」里要保存并启用，否则选不到。

五、部署与运行

pip install fastapi uvicorn pyodbc pandasuvicorn main:app --reload --port 8000# 生产：放公网 + HTTPS（如 Nginx 反代），并加接口鉴权

六、入参 / 出参映射（通用模板）

插件 return 字典的 key = 插件设计里声明的出参 ID，二者必须一致。下面给出映射规律（以 3 个字段示意，实际按你的业务字段数量扩展）：

规律：后端 SQL 用 AS [中文别名] 给每列起名，插件 return 里用 book_info.get("中文别名") 取出，再赋给一个英文/拼音的出参 ID；表单侧把这个出参 ID 绑到对应字段即可。字段越多，按这个规律往下加行就行。

七、可复用检查清单

SQL 用 ? 参数化，未拼接字符串（防注入）
取最新一条用 TOP 1 + ORDER BY 版本 DESC, 序号 DESC（字符串用 TRY_CAST）
返回前把 NaN 转 None
插件代码是顶层代码 + 结尾 return，不是 def main()
插件里未使用 data 作变量名（改用 book_info 之类）
直接引用 triggerConf 读入参，未使用 globals()
后端已部署到公网 HTTPS，且加了接口鉴权
简道云插件已「启用」；入参、出参 ID 与表单字段映射正确
先用 curl 自测后端，再用插件调试工具发真实主键联调

八、小结

这个套路的本质是「云端插件只做转发，重逻辑留在自建后端」：

数据库在内网？交给后端连，插件不碰凭证。
字段多、逻辑复杂？全写在后端 SQL / 代码里，插件只负责传参与回填。
坑基本都在插件代码规范上（data 变量名、globals() 禁用、顶层代码 + return），后端就是普通 FastAPI，该怎么写怎么写。

照着上面的清单和模板，把「主键」「业务字段」替换成你自己的内容，就能快速复制出同类插件。