自建轻量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 均可。
• Python：3.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，内容如下。

fastapi>=0.110 uvicorn[standard]>=0.29 sqlalchemy>=2.0 pydantic>=2.6 pydantic-settings>=2.2 cryptography>=42.0 PyMySQL>=1.1 sqlparse>=0.5.0 requests>=2.31 python-dotenv>=1.0

执行安装命令。

pip install -r requirements.txt

补充解释几个关键库。

• sqlparse用来做 SQL 语句拆分和只读校验，比简单的字符串判断可靠。
• PyMySQL是 Python 连接 MySQL 的驱动。
• cryptography用于连接密码的对称加密。
• APScheduler本文使用 3.x 稳定版本，因为 4.x 的 API 变化比较大。

4. 核心功能实现

4.1 配置文件与数据库初始化

新建.env文件，至少包含下面两个环境变量。

SECRET_KEY=please-change-this-secret-key API_TOKEN=dev-token

SECRET_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_encoding="utf-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(bind=engine, autoflush=False, autocommit=False) class Base(DeclarativeBase): pass def get_db(): db = SessionLocal() try: yield db finally: db.close()

SQLite 的check_same_thread=False是为了让 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_key=True, autoincrement=True) name: Mapped[str] = mapped_column(String(50), unique=True, index=True) host: Mapped[str] = mapped_column(String(200)) port: Mapped[int] = mapped_column(Integer, default=3306) 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, default=True) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) class SavedQuery(Base): __tablename__ = "saved_queries" id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) 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), nullable=True) destination_url: Mapped[Optional[str]] = mapped_column(String(500), nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) class ExecutionLog(Base): __tablename__ = "execution_logs" id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) query_id: Mapped[Optional[int]] = mapped_column( ForeignKey("saved_queries.id"), nullable=True ) connection_id: Mapped[Optional[int]] = mapped_column(nullable=True) status: Mapped[str] = mapped_column(String(20), default="success") detail: Mapped[str] = mapped_column(Text, default="") created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.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