最近陆续有数据团队在讨论迁移方案以前依赖的在线 SQL 编辑器与定时调度工具开始调整商业化策略PopSQL、SeekWell 这一类 SaaS 产品能否长期续费逐渐变成未知数。团队里的 SQL 资产、共享查询、定时报告一旦绑死在某个厂商真遇到“关停或大幅收费调整”时迁移成本非常高。我自己在给团队做替代方案时发现市面上很难找到一套同时满足“轻量协作 SQL 编辑”和“SQL 查询定时调度”的开源平替。与其一个个 SaaS 试不如自建一个最小可用平台把数据库连接管理、查询保存、只读执行、定时调度、结果通知这几件事做扎实业务分析链路就不会断。本文会从需求拆解开始逐步实现一个可运行的自建版本。后端使用 Python FastAPI SQLAlchemy APScheduler前端先做一个最简查询页面。整体代码偏工程化但又不会引入过重的基础设施适合个人开发者、中小团队参考。如果你也在为 PopSQL、SeekWell 关停背景下的工具选型发愁这篇文章应该能帮你建立一个有用的最低可行方案。1. 背景与需求拆解先别急着找替代品1.1 团队为什么依赖 PopSQL、SeekWell 这类工具很多后端团队和数据分析团队都遇到过同一个问题MySQL 客户端、Navicat、DataGrip 虽然好用但是多人协作能力很弱。每个人电脑里保存一份 SQL 文件连数据库地址、账号都靠口头传递等真正需要统一口径时很难维护。PopSQL 这类协作式 SQL 编辑器解决的是“SQL 脚本可以集中保存、多人共享、按权限执行”的问题。它把查询编辑器带到了浏览器里团队成员可以打开同一个查询改完以后其他人立刻能看到最新版本。SeekWell 这类工具则偏向“自动化执行”。它可以把 SQL 查询跑出结果后推到业务群里、发送给指定人员或者定期刷新某个数据集。对于没有完整数据平台的团队来说这类工具填补了“轻量报表自动化”的空白。1.2 关停或调整后最痛的是这三件事当类似 PopSQL、SeekWell 这样的产品出现关闭、并购、商业策略收紧时团队通常面临三个问题。第一SQL 资产可能拿不回来。团队在平台上沉淀的查询脚本、变量定义、注释说明如果没有完整的导出能力很容易丢失。第二定时任务会静默中断。平时每天自动跑的销售日报、库存预警、数据同步一旦 SaaS 停止服务负责跑数的人根本不会第一时间发现。第三权限和连接信息需要重新梳理。SaaS 平台中保存的数据库账号密码、只读权限、可访问库表范围平台关闭后需要运维重新管理。这三个问题如果集中爆发业务侧最直接的感受就是“没人报数了”。所以自建替代品的核心目标不是再造一个炫酷的 BI 系统而是先保住 SQL 的协作能力和定时执行能力。1.3 最小可用替代方案的能力边界一个可落地的替代平台至少要具备四块能力。连接信息管理能录入数据库连接密码不能明文落库。SQL 保存与复用能把常用查询按名称保存后续直接运行。SQL 只读执行返回结果集默认禁止写操作避免误改线上库。定时调度按 cron 表达式周期性执行 SQL并把结果写入审计日志或通知外部系统。在这个基础之上后续可以再加权限系统、多数据源、结果缓存、前端 SQL 编辑器高亮但“四块核心能力”必须先跑通。下文的技术实现就按这四块来展开。2. 技术选型与整体架构设计2.1 为什么选择 FastAPI SQLAlchemy APScheduler自建内部工具时最怕引入一堆复杂组件导致后续没人维护。这套方案选型以简单优先FastAPI开发效率高自带参数校验和 OpenAPI 文档写内部 API 很方便。SQLAlchemy负责平台自身元数据的存储例如连接配置、SQL 脚本、执行日志。APScheduler提供进程内定时任务能力支持 cron 表达式足够覆盖中小团队的周期性跑数场景。cryptography用来加密数据库密码避免明文保存在数据库里。这里的 SQL 执行不直接使用某个数据库客户端而是通过 SQLAlchemy 动态创建数据库连接。也就是说平台自身元数据使用 SQLite 保存执行查询时再按连接配置连接到用户填写的目标数据库。2.2 系统模块与调用关系从调用链路来看整体结构如下。浏览器 / API 客户端 | v FastAPI 路由层身份校验、参数校验 | |--- 读取/保存查询 ----- SQLite 元数据库 | v 查询执行器 QueryExecutor | v MySQL / PostgreSQL / 其他目标数据源 ^ | APScheduler 定时任务按 cron 触发并记录日志需要说明的是APScheduler 在本文中承担的是轻量调度职责。如果未来任务量变大例如几百个定时查询、需要精确到秒级调度、需要失败重试和分布式执行建议把“定时触发器”和“任务执行器”拆开例如引入 Celery 或 RQ。2.3 安全模型只读优先自建 SQL 平台最危险的地方在于“谁来执行 SQL”。如果让用户直接填任意 SQL 并连上有写权限的账号一旦有人误执行DELETE或DROP后果不可挽回。本文的默认安全策略是目标数据库使用只读账号这是最可靠的一层隔离。应用层再做一次 SQL 白名单校验只允许执行SELECT、SHOW、DESCRIBE、EXPLAIN、WITH等只读语句。密码加密存储并且服务端通过 HTTPS 对外提供接口。需要强调的是应用层关键词过滤只是兜底不能替代数据库账号权限。真正生产环境里必须为平台创建独立的只读数据库账号并只授予该平台所需的库表查询权限。3. 环境准备与项目结构3.1 运行环境说明本文代码在以下环境中验证思路操作系统Windows / macOS / Linux 均可。Python3.10 或更高版本。目标数据库MySQL 5.7 或更高版本示例账号为只读账号。平台自身元数据SQLite本地演示零成本。构建工具pip 或 pipenv。实际项目中的 Python 版本、MySQL 版本可能有差异重点参考实现思路版本号需要根据你的环境调整。3.2 项目目录设计为了便于阅读采用模块化目录而不是把所有代码写到一个文件里。sql-collab/ ├── requirements.txt ├── .env ├── main.py ├── app/ │ ├── __init__.py │ ├── config.py │ ├── database.py │ ├── models.py │ ├── security.py │ ├── schemas.py │ ├── executor.py │ ├── scheduler.py │ └── routes.py ├── static/ │ └── index.html其中main.py负责启动 FastAPI 服务app/executor.py是查询执行核心app/scheduler.py是定时任务逻辑static/index.html是简易浏览器页面。3.3 安装依赖创建requirements.txt内容如下。fastapi0.110 uvicorn[standard]0.29 sqlalchemy2.0 pydantic2.6 pydantic-settings2.2 cryptography42.0 PyMySQL1.1 sqlparse0.5.0 requests2.31 python-dotenv1.0执行安装命令。pip install -r requirements.txt补充解释几个关键库。sqlparse用来做 SQL 语句拆分和只读校验比简单的字符串判断可靠。PyMySQL是 Python 连接 MySQL 的驱动。cryptography用于连接密码的对称加密。APScheduler本文使用 3.x 稳定版本因为 4.x 的 API 变化比较大。4. 核心功能实现4.1 配置文件与数据库初始化新建.env文件至少包含下面两个环境变量。SECRET_KEYplease-change-this-secret-key API_TOKENdev-tokenSECRET_KEY必须固定。它会被派生成 Fernet 加解密密钥如果每次启动都随机变化已经保存的数据库密码将无法解密。API_TOKEN是接口访问令牌实际部署时应改成强随机字符串。新建app/config.py。# 文件app/config.py from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config SettingsConfigDict(env_file.env, env_file_encodingutf-8) app_name: str sql-collab secret_key: str please-change-me api_token: str dev-token metadata_db_url: str sqlite:///./meta.db default_row_limit: int 1000 settings Settings()新建app/database.py。# 文件app/database.py from sqlalchemy import create_engine from sqlalchemy.orm import DeclarativeBase, sessionmaker from .config import settings engine create_engine( settings.metadata_db_url, connect_args{check_same_thread: False} if settings.metadata_db_url.startswith(sqlite) else {}, ) SessionLocal sessionmaker(bindengine, autoflushFalse, autocommitFalse) class Base(DeclarativeBase): pass def get_db(): db SessionLocal() try: yield db finally: db.close()SQLite 的check_same_threadFalse是为了让 FastAPI 的线程池能访问同一个 SQLite 文件。生产环境如果元数据量变大建议把metadata_db_url换成 PostgreSQL。4.2 定义元数据模型平台需要记录三张表连接配置、保存的 SQL、执行日志。新建app/models.py。# 文件app/models.py from datetime import datetime from typing import Optional from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text from sqlalchemy.orm import Mapped, mapped_column from .database import Base class ConnectionConfig(Base): __tablename__ connection_configs id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) name: Mapped[str] mapped_column(String(50), uniqueTrue, indexTrue) host: Mapped[str] mapped_column(String(200)) port: Mapped[int] mapped_column(Integer, default3306) username: Mapped[str] mapped_column(String(100)) encrypted_password: Mapped[str] mapped_column(String(512), default) database: Mapped[str] mapped_column(String(100)) read_only: Mapped[bool] mapped_column(Boolean, defaultTrue) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) class SavedQuery(Base): __tablename__ saved_queries id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) title: Mapped[str] mapped_column(String(100)) connection_id: Mapped[int] mapped_column(ForeignKey(connection_configs.id)) sql: Mapped[str] mapped_column(Text) schedule_cron: Mapped[Optional[str]] mapped_column(String(100), nullableTrue) destination_url: Mapped[Optional[str]] mapped_column(String(500), nullableTrue) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) class ExecutionLog(Base): __tablename__ execution_logs id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) query_id: Mapped[Optional[int]] mapped_column( ForeignKey(saved_queries.id), nullableTrue ) connection_id: Mapped[Optional[int]] mapped_column(nullableTrue) status: Mapped[str] mapped_column(String(20), defaultsuccess) detail: Mapped[str] mapped_column(Text, default) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow)字段设计并不复杂。SavedQuery.schedule_cron记录了该条 SQL 的定时规则为空表示不调度。4.3 密码加密与接口认证新建app/security.py。# 文件app/security.py import base64 import hashlib from cryptography.fernet import Fernet from fastapi import Header, HTTPException, status from .config import settings def _derive_key(secret: str) - bytes: digest hashlib.sha256(secret.encode(utf-8)).digest() return base64.urlsafe_b64encode(digest) cipher Fernet(_derive_key(settings.secret_key)) def encrypt_secret(plain: str) - str: return cipher.encrypt(plain.encode(utf-8)).decode(utf-8) def decrypt_secret(token: str) - str: return cipher.decrypt(token.encode(utf-8)).decode 自建轻量SQL协作平台：从PopSQL/SeekWell迁移到FastAPI与APScheduler
最近陆续有数据团队在讨论迁移方案以前依赖的在线 SQL 编辑器与定时调度工具开始调整商业化策略PopSQL、SeekWell 这一类 SaaS 产品能否长期续费逐渐变成未知数。团队里的 SQL 资产、共享查询、定时报告一旦绑死在某个厂商真遇到“关停或大幅收费调整”时迁移成本非常高。我自己在给团队做替代方案时发现市面上很难找到一套同时满足“轻量协作 SQL 编辑”和“SQL 查询定时调度”的开源平替。与其一个个 SaaS 试不如自建一个最小可用平台把数据库连接管理、查询保存、只读执行、定时调度、结果通知这几件事做扎实业务分析链路就不会断。本文会从需求拆解开始逐步实现一个可运行的自建版本。后端使用 Python FastAPI SQLAlchemy APScheduler前端先做一个最简查询页面。整体代码偏工程化但又不会引入过重的基础设施适合个人开发者、中小团队参考。如果你也在为 PopSQL、SeekWell 关停背景下的工具选型发愁这篇文章应该能帮你建立一个有用的最低可行方案。1. 背景与需求拆解先别急着找替代品1.1 团队为什么依赖 PopSQL、SeekWell 这类工具很多后端团队和数据分析团队都遇到过同一个问题MySQL 客户端、Navicat、DataGrip 虽然好用但是多人协作能力很弱。每个人电脑里保存一份 SQL 文件连数据库地址、账号都靠口头传递等真正需要统一口径时很难维护。PopSQL 这类协作式 SQL 编辑器解决的是“SQL 脚本可以集中保存、多人共享、按权限执行”的问题。它把查询编辑器带到了浏览器里团队成员可以打开同一个查询改完以后其他人立刻能看到最新版本。SeekWell 这类工具则偏向“自动化执行”。它可以把 SQL 查询跑出结果后推到业务群里、发送给指定人员或者定期刷新某个数据集。对于没有完整数据平台的团队来说这类工具填补了“轻量报表自动化”的空白。1.2 关停或调整后最痛的是这三件事当类似 PopSQL、SeekWell 这样的产品出现关闭、并购、商业策略收紧时团队通常面临三个问题。第一SQL 资产可能拿不回来。团队在平台上沉淀的查询脚本、变量定义、注释说明如果没有完整的导出能力很容易丢失。第二定时任务会静默中断。平时每天自动跑的销售日报、库存预警、数据同步一旦 SaaS 停止服务负责跑数的人根本不会第一时间发现。第三权限和连接信息需要重新梳理。SaaS 平台中保存的数据库账号密码、只读权限、可访问库表范围平台关闭后需要运维重新管理。这三个问题如果集中爆发业务侧最直接的感受就是“没人报数了”。所以自建替代品的核心目标不是再造一个炫酷的 BI 系统而是先保住 SQL 的协作能力和定时执行能力。1.3 最小可用替代方案的能力边界一个可落地的替代平台至少要具备四块能力。连接信息管理能录入数据库连接密码不能明文落库。SQL 保存与复用能把常用查询按名称保存后续直接运行。SQL 只读执行返回结果集默认禁止写操作避免误改线上库。定时调度按 cron 表达式周期性执行 SQL并把结果写入审计日志或通知外部系统。在这个基础之上后续可以再加权限系统、多数据源、结果缓存、前端 SQL 编辑器高亮但“四块核心能力”必须先跑通。下文的技术实现就按这四块来展开。2. 技术选型与整体架构设计2.1 为什么选择 FastAPI SQLAlchemy APScheduler自建内部工具时最怕引入一堆复杂组件导致后续没人维护。这套方案选型以简单优先FastAPI开发效率高自带参数校验和 OpenAPI 文档写内部 API 很方便。SQLAlchemy负责平台自身元数据的存储例如连接配置、SQL 脚本、执行日志。APScheduler提供进程内定时任务能力支持 cron 表达式足够覆盖中小团队的周期性跑数场景。cryptography用来加密数据库密码避免明文保存在数据库里。这里的 SQL 执行不直接使用某个数据库客户端而是通过 SQLAlchemy 动态创建数据库连接。也就是说平台自身元数据使用 SQLite 保存执行查询时再按连接配置连接到用户填写的目标数据库。2.2 系统模块与调用关系从调用链路来看整体结构如下。浏览器 / API 客户端 | v FastAPI 路由层身份校验、参数校验 | |--- 读取/保存查询 ----- SQLite 元数据库 | v 查询执行器 QueryExecutor | v MySQL / PostgreSQL / 其他目标数据源 ^ | APScheduler 定时任务按 cron 触发并记录日志需要说明的是APScheduler 在本文中承担的是轻量调度职责。如果未来任务量变大例如几百个定时查询、需要精确到秒级调度、需要失败重试和分布式执行建议把“定时触发器”和“任务执行器”拆开例如引入 Celery 或 RQ。2.3 安全模型只读优先自建 SQL 平台最危险的地方在于“谁来执行 SQL”。如果让用户直接填任意 SQL 并连上有写权限的账号一旦有人误执行DELETE或DROP后果不可挽回。本文的默认安全策略是目标数据库使用只读账号这是最可靠的一层隔离。应用层再做一次 SQL 白名单校验只允许执行SELECT、SHOW、DESCRIBE、EXPLAIN、WITH等只读语句。密码加密存储并且服务端通过 HTTPS 对外提供接口。需要强调的是应用层关键词过滤只是兜底不能替代数据库账号权限。真正生产环境里必须为平台创建独立的只读数据库账号并只授予该平台所需的库表查询权限。3. 环境准备与项目结构3.1 运行环境说明本文代码在以下环境中验证思路操作系统Windows / macOS / Linux 均可。Python3.10 或更高版本。目标数据库MySQL 5.7 或更高版本示例账号为只读账号。平台自身元数据SQLite本地演示零成本。构建工具pip 或 pipenv。实际项目中的 Python 版本、MySQL 版本可能有差异重点参考实现思路版本号需要根据你的环境调整。3.2 项目目录设计为了便于阅读采用模块化目录而不是把所有代码写到一个文件里。sql-collab/ ├── requirements.txt ├── .env ├── main.py ├── app/ │ ├── __init__.py │ ├── config.py │ ├── database.py │ ├── models.py │ ├── security.py │ ├── schemas.py │ ├── executor.py │ ├── scheduler.py │ └── routes.py ├── static/ │ └── index.html其中main.py负责启动 FastAPI 服务app/executor.py是查询执行核心app/scheduler.py是定时任务逻辑static/index.html是简易浏览器页面。3.3 安装依赖创建requirements.txt内容如下。fastapi0.110 uvicorn[standard]0.29 sqlalchemy2.0 pydantic2.6 pydantic-settings2.2 cryptography42.0 PyMySQL1.1 sqlparse0.5.0 requests2.31 python-dotenv1.0执行安装命令。pip install -r requirements.txt补充解释几个关键库。sqlparse用来做 SQL 语句拆分和只读校验比简单的字符串判断可靠。PyMySQL是 Python 连接 MySQL 的驱动。cryptography用于连接密码的对称加密。APScheduler本文使用 3.x 稳定版本因为 4.x 的 API 变化比较大。4. 核心功能实现4.1 配置文件与数据库初始化新建.env文件至少包含下面两个环境变量。SECRET_KEYplease-change-this-secret-key API_TOKENdev-tokenSECRET_KEY必须固定。它会被派生成 Fernet 加解密密钥如果每次启动都随机变化已经保存的数据库密码将无法解密。API_TOKEN是接口访问令牌实际部署时应改成强随机字符串。新建app/config.py。# 文件app/config.py from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config SettingsConfigDict(env_file.env, env_file_encodingutf-8) app_name: str sql-collab secret_key: str please-change-me api_token: str dev-token metadata_db_url: str sqlite:///./meta.db default_row_limit: int 1000 settings Settings()新建app/database.py。# 文件app/database.py from sqlalchemy import create_engine from sqlalchemy.orm import DeclarativeBase, sessionmaker from .config import settings engine create_engine( settings.metadata_db_url, connect_args{check_same_thread: False} if settings.metadata_db_url.startswith(sqlite) else {}, ) SessionLocal sessionmaker(bindengine, autoflushFalse, autocommitFalse) class Base(DeclarativeBase): pass def get_db(): db SessionLocal() try: yield db finally: db.close()SQLite 的check_same_threadFalse是为了让 FastAPI 的线程池能访问同一个 SQLite 文件。生产环境如果元数据量变大建议把metadata_db_url换成 PostgreSQL。4.2 定义元数据模型平台需要记录三张表连接配置、保存的 SQL、执行日志。新建app/models.py。# 文件app/models.py from datetime import datetime from typing import Optional from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text from sqlalchemy.orm import Mapped, mapped_column from .database import Base class ConnectionConfig(Base): __tablename__ connection_configs id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) name: Mapped[str] mapped_column(String(50), uniqueTrue, indexTrue) host: Mapped[str] mapped_column(String(200)) port: Mapped[int] mapped_column(Integer, default3306) username: Mapped[str] mapped_column(String(100)) encrypted_password: Mapped[str] mapped_column(String(512), default) database: Mapped[str] mapped_column(String(100)) read_only: Mapped[bool] mapped_column(Boolean, defaultTrue) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) class SavedQuery(Base): __tablename__ saved_queries id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) title: Mapped[str] mapped_column(String(100)) connection_id: Mapped[int] mapped_column(ForeignKey(connection_configs.id)) sql: Mapped[str] mapped_column(Text) schedule_cron: Mapped[Optional[str]] mapped_column(String(100), nullableTrue) destination_url: Mapped[Optional[str]] mapped_column(String(500), nullableTrue) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) class ExecutionLog(Base): __tablename__ execution_logs id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) query_id: Mapped[Optional[int]] mapped_column( ForeignKey(saved_queries.id), nullableTrue ) connection_id: Mapped[Optional[int]] mapped_column(nullableTrue) status: Mapped[str] mapped_column(String(20), defaultsuccess) detail: Mapped[str] mapped_column(Text, default) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow)字段设计并不复杂。SavedQuery.schedule_cron记录了该条 SQL 的定时规则为空表示不调度。4.3 密码加密与接口认证新建app/security.py。# 文件app/security.py import base64 import hashlib from cryptography.fernet import Fernet from fastapi import Header, HTTPException, status from .config import settings def _derive_key(secret: str) - bytes: digest hashlib.sha256(secret.encode(utf-8)).digest() return base64.urlsafe_b64encode(digest) cipher Fernet(_derive_key(settings.secret_key)) def encrypt_secret(plain: str) - str: return cipher.encrypt(plain.encode(utf-8)).decode(utf-8) def decrypt_secret(token: str) - str: return cipher.decrypt(token.encode(utf-8)).decode