PopSQL 和 SeekWell 陆续传出停运/调整的消息时很多团队的数据库查询页面还停留在上一家公司提供的稳定期内。表面上这只是换一个工具的问题实际上团队积累的几百条查询、固定的连接配置、定时输出结果的约定、新成员上手时的共享入口都锁在别人家的服务里。一旦产品关停这些资产要么导出不顺要么语义已经迁不回本地整个数据协作的节奏会被打断。这类工具往往不是纯数据库客户端而是“查询协作平台”。它们覆盖从连库、写 SQL、保存结果、分享链接到定时把查询结果推向其他系统的完整链路。问题在于大多数团队把查询代码当代码资产管理却把“执行这些查询的环境约定”交给了平台托管。平台停止提供服务时损失最重的不是编辑器本身而是围绕 SQL 形成的知识库和协作路径。这篇文章想聊的不只是某个产品的替代方案而是这类工具背后的核心模型一个可自建、可扩展、能接回团队原有数据库权限体系的查询工作台应该由哪些部分组成。文章会先拆解 SQL 协作工具的价值点再给出一套最小可行的自建方案包含数据模型设计、后端执行引擎、Web 交互基础代码以及连库验证、权限边界和后续扩展建议。无论你最后是选择商用替代品还是临时自建过渡这份架构梳理都能帮助你把“查询资产”和“执行入口”解耦避免下一次平台变动时再次被动。1. 工具停服真正暴露的问题是什么在分析自建之前需要先把“工具停服意味着什么”看清楚。如果只是一个数据库 IDE 不能用了用户的损失相对可控因为本地还会保留一些历史连接配置更换客户端成本不难接受。但 PopSQL 这类产品在团队里的定位往往已经超过了单机 IDE 的范畴。一旦团队成员把查询都放在平台上平台就变成了三个角色的集合体。第一它是查询仓库。每个人保存的查询、团队分组、收藏、标签、最近运行版本都已经平台化。本地可能只留下零散文件很多查询是现写现跑从来没有完整导出过。第二它是连接入口。这类平台一般不会把数据库账号直接分发给每个成员而是由管理员在系统里维护连接。成员只需要在界面选择数据源。结果是很多人根本不知道生产库的内网地址、账号权限、连接限制具体是什么。数据库连接信息作为敏感配置被收在平台侧工具一停团队连“访问路径”都变得模糊。第三它是自动化调度节点。很多团队会把常用 SQL 做成定时任务查询完成后自动推送到群、表格或看板。这个能力看起来是小功能实际上承担了日常数据协作的很多固定仪式早上看销售日报、定期同步渠道数据、监控异常订单。平台停运后这些自动化任务全部失效。所以从工程视角看停服暴露的并不仅仅是编辑器从“可用”变成“不可用”而是团队把自己的数据目录、连接语义和调度任务都绑定在了外部服务的生命周期上。工具稳定时这是效率红利工具停运时这就是迁移成本。这也解释了为什么“自建替代”这件事会成为一个值得认真评估的选项。自建并不一定比商业工具更完善但它能保证两件事数据资产留在自己仓库里连接配置由自己控制。对一个已经有数据库运维能力的团队来说这个价值往往超过界面、交互层面的差距。2. 替代品需要覆盖的核心能力与合理边界真正动手写代码之前建议先做一次功能清单剥离。不要试图模仿商业工具的全部功能而是要把团队日常用到的高频能力抽出来范围控制在“能替代旧工具 80% 的日常”。一个最小可用的 SQL 协作工作台通常由三部分构成查询编辑器、保存与分享、数据源管理。再加上一个可选的调度模块就能覆盖大多数查询类协作场景。能力模块核心用户需求停服后最难恢复的部分自建优先级数据源管理管理员集中配置数据库连接成员不需要知道敏感密码多环境的连接约定、账号权限关系高SQL 编辑器快速写查询、格式化、执行、查看结果常用查询的历史记录和片段高查询保存与分享把查询保存下来给同事一个链接就能复现结果团队级查询资产、命名规范高结果查看与导出后端返回结果前端支持翻页和导出特定查询的固定结果口径中权限与审计谁能看哪个数据源、谁执行过什么查询操作记录和敏感查询的可追溯性中定时调度通知查询结果定时推送到群、邮件或外部系统依赖固定任务数据消费的习惯低可作为二期这个表里最关键的是数据源管理和查询保存。前者让团队成员不需要接触数据库明文账号后者让团队知识资产留存下来。两者都解决的是“协作配置”问题而不是 SQL 语法问题。自建的切入边界也应该放在这两个地方而不是重新打磨一个强到不行的编辑器。还需要明确一个边界SQL 工作台不等于 BI 报表系统。BI 工具的核心是预置数据模型、图表、仪表盘和权限矩阵SQL 工作台的核心是“能查、能存、能分享”。两者有重叠但对自建项目来说一旦加入大量 BI 图表设计工程量会迅速膨胀。更务实的路线是先把查询和结果跑通图表、看板、自动推送放到后面的扩展阶段。3. 自建的整体架构查询服务而不是桌面客户端先给结论这类工具最适合的形态不是本地桌面客户端而是一个轻量 Web 服务 浏览器前端。原因是团队协作天然需要中心化保存和共享。本地客户端再强大也无法解决“同事点开链接看到同一份查询结果”的问题。从架构上可以拆成两个核心平面。控制平面保存团队查询合作的元数据。它记录的问题包括谁保存了这条查询查询属于哪个项目对应哪个数据源最近执行耗时是多少。这个平面不需要接触真实业务数据数据量也不大用一个 PostgreSQL、MySQL 甚至 SQLite 就可以承载。开发初期最容易被忽视却是把“单机脚本”升级为“团队工具”的关键。数据平面负责真正执行 SQL。它对接团队已有的业务数据库提供执行入口并返回结果。这里最容易出现的误区是直接让每个团队成员用自己的本机 Navicat 连库然后手动同步 SQL 文本。那样做并没有解决问题连接信息会在聊天软件里到处飞账号权限也完全不可控。数据平面应该由服务端统一持有连接成员只能提交 SQL不能直接拿连接串。把这套思想落到代码上一个典型的服务端结构可以设计成如下形式workbench/ ├── app/ │ ├── main.py # FastAPI 入口注册路由 │ ├── models.py # 查询保存、运行记录等 ORM 模型 │ ├── query_engine.py # 核心执行引擎封装 SQLAlchemy 连接池 │ ├── data_sources.py # 数据源配置文件加载器 │ └── templates/ │ └── index.html # 简化版前端页面 ├── data_sources.yaml # 数据源连接配置 └── requirements.txt服务端选择 Python FastAPI主要是因为搭建速度快、异步能力足够同时 SQLAlchemy 对主流数据库支持很成熟。整个 Web 层其实只是一个薄壳核心逻辑都集中在两块一块负责保存查询元数据一块负责把 SQL 交给正确的数据库引擎执行。前端只要负责编辑 SQL 和展示结果不需要直接连接数据库这样就规避了在浏览器里分发数据库客户端和账号密码的安全问题。这个架构真正的优势是简单清楚前端只跟后端通信配置和数据源都收口在后端。未来要增加调度任务可以直接复用后端的执行引擎不需要改变前端结构。4. 环境准备与技术选型在进入代码之前先确认一下环境。示例代码会用 Python 3.10 和 FastAPI数据库元数据临时用 SQLite 保存这是为了最小化部署成本。如果你要把方案用于生产模型层可以平滑切换到 PostgreSQLSQLAlchemy 对这两类数据库兼容都很好。建议先创建一个干净的虚拟环境避免和系统 Python 包互相污染。mkdir -p sql-workbench/app/templates cd sql-workbench python -m venv .venv source .venv/bin/activate pip install fastapi uvicorn[standard] sqlalchemy2.0 \ pydantic2.0 pyyaml如果你希望后续执行目标数据库的能力更强可以按需安装对应驱动。这里不限制具体版本建议以项目实际依赖为准。比如你的团队主要用 PostgreSQL就可以加上pip install psycopg[binary]如果用 MySQL则安装pip install pymysql本文后面的执行引擎基于 SQLAlchemy 2.0 风格编写能兼容 PostgreSQL、MySQL、SQLite 等常见关系型数据库。因为示例使用 SQLite 做演示不需要额外启动数据库容器跑通之后再把连接配置替换成目标数据库即可。为了验证效果需要先准备 Demo 数据。我们会在本地生成一个临时 SQLite 文件里面放一张销售订单表方便执行查询后看到真实结果。5. 数据模型设计查询保存与运行记录第一步先定义控制平面的数据结构。一个 SQL 协作工具最核心的表是保存查询表。它保存的不只是 SQL 文本还要记录数据源名称、创建人、创建时间这样团队成员才能按项目维度浏览历史资产。文件路径app/models.pyfrom datetime import datetime from sqlalchemy import BigInteger, DateTime, String, Text, func from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column class Base(DeclarativeBase): pass class SavedQuery(Base): __tablename__ saved_query id: Mapped[int] mapped_column(BigInteger, primary_keyTrue) name: Mapped[str] mapped_column(String(200), nullableFalse) project: Mapped[str] mapped_column(String(100), nullableFalse, defaultdefault) datasource: Mapped[str] mapped_column(String(100), nullableFalse) owner: Mapped[str] mapped_column(String(100), nullableFalse) sql_text: Mapped[str] mapped_column(Text, nullableFalse) created_at: Mapped[datetime] mapped_column( DateTime(timezoneTrue), server_defaultfunc.now() ) updated_at: Mapped[datetime] mapped_column( DateTime(timezoneTrue), server_defaultfunc.now(), onupdatefunc.now() )这个模型里datasource字段并不直接保存数据库连接串而是指向一个逻辑名称例如primary_ro或analytics_readonly。真正的连接信息放在数据源配置文件里由服务端统一维护。这样做有两点好处团队成员可以在不知道账号密码的情况下使用数据源管理员要更换数据库地址时只需要改配置文件不需要让所有成员更新连接信息。再增加一张运行记录表用于记录谁在什么时间执行过哪条查询。这张表是审计的基础也能帮助团队分析哪些查询是高频查询为后续优化索引或缓存提供依据。class QueryRunRecord(Base): __tablename__ query_run_record id: Mapped[int] mapped_column(BigInteger, primary_keyTrue) saved_query_id: Mapped[int] mapped_column(BigInteger, nullableTrue) datasource: Mapped[str] mapped_column(String(100), nullableFalse) owner: Mapped[str] mapped_column(String(100), nullableFalse) status: Mapped[str] mapped_column(String(20), nullableFalse, defaultsuccess) elapsed_ms: Mapped[int] mapped_column(BigInteger, nullableTrue) error_message: Mapped[str] mapped_column(Text, nullableTrue) created_at: Mapped[datetime] mapped_column( DateTime(timezoneTrue), server_defaultfunc.now() )在演示项目中这个模型可以只用于记录日志不参与查询执行验证。生产环境还可以增加项目分组表、用户权限关联表、敏感字段标记等但第一步先把核心链路跑通。6. 数据源配置与查询执行引擎数据源配置文件负责保存“连接到哪个地址、使用什么只读账号”。这里刻意把 URL 放在 YAML 里是为了让演示直观。生产环境建议用环境变量、密钥管理服务或配置中心替换不要把包含数据库账号的内容提交到 Git。文件路径data_sources.yamldatasources: demo_sqlite: display_name: 演示数据源 driver: sqlite url: sqlitepysqlite:////tmp/sql-workbench-demo.db default_limit: 500后续要接入真实 PostgreSQL只需要增加这样一个数据源节点primary_ro: display_name: 主业务库只读 driver: postgresql # 生产环境强烈建议使用只读账号而不要依赖程序拦截 url: postgresqlpsycopg://readonly_user:password192.168.10.20:5432/app_db default_limit: 1000关键在于“只读账号”这几个字。靠后端解析 SQL 是拦不住所有写操作的最稳妥的防线是数据库账号本身没有 INSERT 和 UPDATE 权限。业务库使用独立低权限账号是审计安全的第一原则。然后写一个配置加载器和一个执行引擎。配置加载器负责把 YAML 转成 Python 字典。文件路径app/data_sources.pyfrom pathlib import Path import yaml _config None def load_config() - dict: global _config if _config is None: _config yaml.safe_load(Path(data_sources.yaml).read_text(encodingutf-8)) return _config def get_datasource(name: str) - dict: sources load_config()[datasources] if name not in sources: raise KeyError(f数据源 {name} 不存在) return sources[name]执行引擎封装 SQLAlchemy 的 Engine。同一个数据源只需要创建一次 Engine由连接池复用不要每次执行都重新创建连接。文件路径app/query_engine.pyimport time from typing import Any from sqlalchemy import create_engine, text from sqlalchemy.engine import Engine from app.data_sources import get_datasource _engines: dict[str, Engine] {} def _get_engine(name: str) - Engine: if name not in _engines: source get_datasource(name) _engines[name] create_engine( source[url], pool_pre_pingTrue, pool_recycle3600, ) return _engines[name] def _to_jsonable(value: Any) - Any: if value is None: return None if isinstance(value, (str, int, float, bool)): return value return str(value) def run_sql(datasource: str, sql: str, limit: int 500) - dict: sql sql.strip() if not sql: raise ValueError(SQL 不能为空) source get_datasource(datasource) if limit source.get(default_limit, 1000): limit source.get(default_limit, 1000) engine _get_engine(datasource) start time.time() # 这里使用 with 块管理连接和事务边界。 # 演示环境的数据源账号应配置为只读生产环境不要依赖程序拦截写操作。 with engine.connect() as connection: result connection.execute(text(sql)) columns list(result.keys()) if result.returns_rows else [] rows [] row_count 0 if result.returns_rows: for row in result: if row_count limit: break rows.append([_to_jsonable(value) for value in row]) row_count 1 elapsed_ms int((time.time() - start) * 1000) return { datasource: datasource, columns: columns, rows: rows, row_count: row_count, truncated: row_count limit, elapsed_ms: elapsed_ms, }这段代码的关键点是先用 strip 去掉 SQL 首尾空白再交给数据库执行读取结果时用limit兜底防止一次查询把后端内存打满所有返回值都尽量转成 JSON 可序列化类型。elapsed_ms可以在前端直观展示查询耗时帮助团队发现慢查询。如果执行失败可以在这里捕获异常并向上抛出由 API 层转换成 HTTP 错误响应。捕获异常时不要把数据库底层错误原样抛给前端生产环境可以只返回“查询执行失败”具体原因写入服务端日志。7. FastAPI 接口层与最小前端页面有了执行引擎还需要 API 层把 HTTP 请求映射到引擎。设计两个主要接口POST /api/run传入数据源名称和 SQL立即执行并返回结果。POST /api/saved_queries把常用查询保存到元数据库。文件路径app/main.pyfrom contextlib import asynccontextmanager from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from sqlalchemy import create_engine from sqlalchemy.orm import Session from app import models from app.query_engine import run_sql META_DATABASE_URL sqlitepysqlite:////tmp/sql-workbench-meta.db meta_engine create_engine(META_DATABASE_URL, connect_args{check_same_thread: False}) asynccontextmanager async def lifespan(app: FastAPI): models.Base.metadata.create_all(meta_engine) yield app FastAPI(titleSQL Workbench Demo, lifespanlifespan) class RunRequest(BaseModel): datasource: str sql: str limit: int Field(default200, ge1, le5000) class SaveRequest(BaseModel): name: str project: str default datasource: str owner: str tester sql_text: str app.post(/api/run) def api_run(req: RunRequest): try: return run_sql(req.datasource, req.sql, req.limit) except KeyError as exc: raise HTTPException(status_code400, detailstr(exc)) except Exception as exc: # 生产环境建议只记录日志返回“查询执行失败” raise HTTPException(status_code400, detailstr(exc)) app.post(/api/saved_queries) def api_save_query(req: SaveRequest): with Session(meta_engine) as session: query models.SavedQuery( namereq.name, projectreq.project, datasourcereq.datasource, ownerreq.owner, sql_textreq.sql_text, ) session.add(query) session.commit() return {id: query.id, message: 保存成功}这里为了演示省去了用户认证。生产环境一定要在路由上增加认证中间件并从登录态中获取owner不能让用户随意传一个 owner 字段。否则任何人把 owner 改成别人就能伪造查询记录归属。前端的职责被刻意压到最小提供一个数据源输入、一个 SQL 编辑区域和一个结果展示区。点击“运行”后把 SQL 发给后端并把返回结果渲染成表格。文件路径app/templates/index.html!DOCTYPE html html langzh-CN head meta charsetUTF-8 titleSQL Workbench Demo/title style body { font-family: sans-serif; margin: 24px; } textarea { width: 100%; height: 140px; font-family: monospace; } button { margin: 8px 0; padding: 8px 16px; } table { border-collapse: collapse; margin-top: 16px; } th, td { border: 1px solid #ddd; padding: 6px 10px; font-size: 14px; } .meta { color: #666; font-size: 13px; margin-top: 8px; } /style /head body h2SQL 查询工作台最小实现/h2 div label数据源/label input iddatasource valuedemo_sqlite / /div textarea idsqlTextSELECT * FROM orders LIMIT 20;/textarea div button onclickrunQuery()运行查询/button /div div idmeta classmeta/div div idresultArea/div script async function runQuery() { const payload { datasource: document.getElementById(datasource).value.trim(), sql: document.getElementById(sqlText).value.trim(), limit: 200 }; const meta document.getElementById(meta); const resultArea document.getElementById(resultArea); meta.textContent 执行中...; resultArea.innerHTML ; const resp await fetch(/api/run, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify(payload) }); if (!resp.ok) { const err await resp.text(); meta.textContent 执行失败 err; return; } const data await resp.json(); meta.textContent 行数 data.row_count 耗时 data.elapsed_ms ms (data.truncated ? 结果已截断 : ); const table document.createElement(table); const headRow document.createElement(tr); data.columns.forEach(col { const th document.createElement(th); th.textContent col; headRow.appendChild(th); }); table.appendChild(headRow); data.rows.forEach(row { const tr document.createElement(tr); row.forEach(value { const td document.createElement(td); td.textContent value; tr.appendChild(td); }); table.appendChild(tr); }); resultArea.appendChild(table); } /script /body /html创建完静态 HTML 之后需要在 FastAPI 中注册模板目录或静态文件目录。最简单的方式是在 main 里添加挂载pip install jinja2然后在main.py中加入from fastapi.responses import HTMLResponse from fastapi.templating import Jinja2Templates from fastapi import Request templates Jinja2Templates(directoryapp/templates) app.get(/, response_classHTMLResponse) def home(request: Request): return templates.TemplateResponse(request, index.html)不同 FastAPI 版本的 TemplateResponse 参数顺序略有差异自己确认 IDE 提示即可。8. 运行验证与问题排查为了能看到执行结果先准备一张示例表和几条订单数据。我们之间不需要引入测试框架直接用一个 Python 脚本向/tmp/sql-workbench-demo.db写入数据。在项目根目录新建prepare_demo.pyfrom sqlalchemy import create_engine, text engine create_engine(sqlitepysqlite:////tmp/sql-workbench-demo.db) with engine.begin() as conn: conn.execute(text(DROP TABLE IF EXISTS orders)) conn.execute(text( CREATE TABLE orders ( id INTEGER PRIMARY KEY, order_no VARCHAR(32), customer VARCHAR(64), amount NUMERIC(10, 2), created_at DATETIME ) )) conn.execute(text( INSERT INTO orders (order_no, customer, amount, created_at) VALUES (SO-001, 张三, 199.00, 2025-01-05 10:20:00), (SO-002, 李四, 399.50, 2025-01-06 11:30:00), (SO-003, 王五, 89.90, 2025-01-07 09:15:00), (SO-004, 张三, 328.00, 2025-01-08 14:45:00) )) print(demo data ready)运行命令python prepare_demo.py python -m uvicorn app.main:app --reload在浏览器打开http://127.0.0.1:8000页面会加载index.html。默认的查询是SELECT * FROM orders LIMIT 20;点击“运行查询”预期出现一个四列的数据表并显示类似行数4耗时2ms如果希望验证保存查询接口可以使用 curlcurl -s -X POST http://127.0.0.1:8000/api/saved_queries \ -H Content-Type: application/json \ -d { name: 最近订单, project: sales, datasource: demo_sqlite, owner: tester, sql_text: SELECT * FROM orders ORDER BY id DESC LIMIT 10; }预期返回{id:1,message:保存成功}如果运行失败可以按下面顺序排查问题现象可能原因排查方式解决方案启动报 module 找不到依赖没装全确认是否激活虚拟环境执行pip list查看依赖安装 requirements.txt 中的依赖页面显示 404FastAPI 路由没注册检查 main.py 中是否挂载模板目录是否定义了/路由参考第 7 节补充静态页面路由执行查询时报数据源不存在YAML 里的名字写错查看 data_sources.yaml比对datasource字段统一成小写英文名称返回成功但接口报 500目标数据库驱动缺失查看 uvicorn 控制台日志安装对应数据库驱动比如pip install pymysql查询结果为空但 SQL 正确limit 设置过小或数据源默认 limit 过小检查返回对象里的truncated字段调大请求 limit 或数据源 default_limit这套最小实现已经覆盖了连接管理、查询执行和查询保存三个核心环节。你可以把 SQLite 数据源替换成真实的业务数据库只要数据源连接账号配置为只读就可以让团队试跑。9. 安全边界与工程建议自建 SQL 工作台最容易被低估的是安全边界。很多初级实现把所有精力放在“能跑查询”上却忽略了查询工具本身就是一种数据库入口。如果权限设计不当等于给团队内部开放了一个弱管控的数据通道。以下几个原则需要特别强调。第一数据库账号必须是低权限只读账号。不要使用 DBA 账号去配置工作台数据源。即使前端没有提供写按钮攻击者也可以绕过页面直接调用接口填入UPDATE、DELETE甚至TRUNCATE语句。靠业务层过滤 SQL 并不安全数据库账号层面的权限收敛才是第一道防线。第二必须有认证和授权。本文示例由于聚焦核心流程没有加登录。生产环境需要接入公司的统一登录、OAuth 或 SSO并且每个用户的数据源权限应该单独配置。不是所有成员都能查生产库也不是所有成员都能导出全量数据。第三查询任务要做超时控制。线上数据库一旦被一个慢查询拖住影响面会很大。常见的做法是在数据库端设置 statement timeout并在后端执行连接时配置连接超时。如果使用 PostgreSQL可以用ALTER ROLE为只读账号设置单语句超时或者在连接字符串里启用超时参数。SQLite 本地演示不涉及这个问题真实数据源必须处理。第四结果集大小要有限制。返回 100 万行到前端浏览器不仅没有意义还会拖垮服务端内存。建议默认返回 200 行允许用户显式调整到 5000 行以内。导出大结果集应该走异步任务不要占用同步 HTTP 请求。第五日志要结构化。执行人、执行时间、数据源、SQL 全文、执行耗时、返回行数、错误信息都应该记录。SQL 文本可能包含敏感查询条件但作为审计信息通常还是需要留存。至少需要记录执行人和执行时间敏感 SQL 全文可以按公司安全策略决定是否落库。从工程演进角度看完成本文这个最小版本之后可以作为二期扩展的方向包括查询分享链接、项目分组与权限、操作审计页面、参数化查询、历史执行记录对比、定时调度与群通知。定时调度实际上并不难只需要在服务端增加一个后台线程或调度框架定期从saved_query表里读取标记为定时执行的任务调用同一个run_sql函数再把结果推送到 Webhook。这个扩展方案能复用现有核心引擎不会推翻原来的架构。最后想提醒一点不要为了自建而自建。如果团队已经有人力维护商业工具订阅且没有强数据合规诉求更换为另一个成熟的商业产品往往更划算。自建
