SQLite + Flask 的正确打开方式：从"每个请求开 12 次连接"到"最多 1 次"

一、问题在哪

venxine.vip 有大量的页面需要查 SQLite：

/system/ 查 sessions 总数、token 总量、uptime
/token/ 按模型、按日期、按周汇总 token 用量
/api/token-stats 查 30 天 + 7 天 + 全期三期数据
/api/daily-tokens 查按月聚合
/cron/ 查 cron 任务统计

每个路由函数里都散落着类似的代码：

@app.route("/api/token-stats")
def token_stats():
    conn = sqlite3.connect(db_path)   # 第 1 次
    # ... 30 天统计 ...
    conn.close()

    conn7 = sqlite3.connect(db_path)  # 第 2 次
    # ... 7 天统计 ...
    conn7.close()

    conn_all = sqlite3.connect(db_path)  # 第 3 次
    # ... 全期统计 ...
    conn_all.close()

还有后续的辅助查询：

    conn = sqlite3.connect(db_path)   # 第 4 次
    # ...
    conn.close()

    conn2 = sqlite3.connect(db_path)  # 第 5 次
    # ...
    conn2.close()

统计后发现：复杂路由每次请求打开 5-12 次 SQLite 连接。

二、为什么这是个问题

每次 sqlite3.connect() 都要：

打开文件句柄
读取 SQLite header
初始化内存缓存
检查 WAL/journal 状态

单次很快（~1ms），但 12 次就是 12ms，并发场景下会被放大。

更隐蔽的是文件句柄泄漏风险。如果某个中间步骤抛了异常，conn.close() 没执行到，SQLite 文件锁就残留在内存里，下次打开可能报 database is locked。

三、教科书方案：teardown_appcontext（为什么我最后没用它）

Flask 官方文档推荐的模式：

from flask import g

def get_db():
    if "db" not in g:
        g.db = sqlite3.connect(DATABASE)
        g.db.row_factory = sqlite3.Row
    return g.db

@app.teardown_appcontext
def close_db(exception):
    db = g.pop("db", None)
    if db is not None:
        db.close()

这个方案的问题在于 teardown 时机不可控。这个项目的一些路由函数内部会多次读写数据库，每次读完就关闭，然后再打开。但 teardown_appcontext 不知道这一点：它在请求结束时还会尝试 pop 和 close 一次已经关闭的连接。

核心矛盾：teardown_appcontext 假设每个请求只用一个连接，结束时关闭。但这个项目的路由函数内有多次"打开、查询、关闭、再打开"的循环。

四、真方案：检查 -> 创建 -> 显式关闭

from flask import g as _g

def _get_state_db():
    """获取 state.db 连接，请求内复用"""
    if "_state_db" not in _g:
        _g._state_db = sqlite3.connect(
            os.path.expanduser("~/.hermes/state.db"))
        _g._state_db.row_factory = sqlite3.Row
    return _g._state_db

def _close_state_db():
    """关闭并清理连接"""
    db = _g.pop("_state_db", None)
    if db is not None:
        db.close()

用法：

@app.route("/api/token-stats")
def token_stats():
    # 第一次调用 -> 创建连接
    conn = _get_state_db()
    # 查 30 天数据
    _close_state_db()   # 用完关闭

    # 第二次调用 -> 重新创建（因为上一步 pop 掉了）
    conn7 = _get_state_db()
    # 查 7 天数据
    _close_state_db()

    # 第三次调用 -> 再次创建
    conn_all = _get_state_db()
    # 查全期数据
    _close_state_db()

关键区别：每次关闭后 g.pop() 清掉了 key，下次 _get_state_db() 发现 key 不存在就重新创建。不会出现"返回已关闭的连接"的问题。

先说句公道话：教科书里"一个请求一个连接、teardown 收尾"才是最干净的写法，多数项目就该那么做。我这里之所以要手动反复开关，是因为几个老路由把十来次查询堆在了同一个函数里，短期内不想动它们的结构。下面这套显式管理是给这种历史包袱做的务实兜底，不是推荐范式。等哪天把路由拆干净了，回到一请求一连接会更省心。

五、为什么复用而不是一直不关

你可能会问：既然是同一个数据库，为什么不保持一个长连接？

答案：这个项目用了两个 SQLite 数据库，查询完后没理由一直持有连接。

# state.db -> Token 统计、Session 记录
_get_state_db()  # 创建
# ... 查询 ...
_close_state_db()  # 关闭

# memory.db -> Agent 记忆查询
_get_memory_db()  # 创建
# ... 查询 ...
_close_memory_db()  # 关闭

每个请求只在这两个数据库之间切换一次，关闭之后释放给其他请求。SQLite 本身是单写多读，短连接不会成为瓶颈。

六、对比：优化前后的调用模式

以一个典型路由的请求为例：

优化前：

sqlite3.connect("state.db")  -> #1
conn.close()
sqlite3.connect("state.db")  -> #2
conn2.close()
sqlite3.connect("memory.db") -> #3
conn.close()
--------------------------------
共 3 次直接 connect，关闭逻辑分散

优化后：

_get_state_db()      -> 检查 g，不存在则创建
  #1 查询
_close_state_db()    -> g.pop() + close()

_get_state_db()      -> 检查 g，不存在则创建
  #2 查询
_close_state_db()

_get_memory_db()     -> 检查 g，不存在则创建
  #3 查询
_close_state_db()
--------------------------------
逻辑上还是 3 次 create/close
但不会泄漏、不会复用已关闭连接

真正受益的场景是像 /api/token-stats 这种 12 次调用的大路由，显式管理让每个连接的生命周期完全可控。

七、为什么不引入 SQLAlchemy

经常有人问：为什么不用 ORM？

答案很简单：这个项目不需要。

场景	SQLAlchemy 收益	代价
3 个表、简单查询	几乎没有	依赖体积 + 学习成本
多数据库支持	0（只有 SQLite）	ORM 抽象层开销
复杂 JOIN	没有	—
连接池管理	有	自带的 pool 也够用

SQLite + Flask g + 显式管理是当前规模的最佳搭配。等到查询复杂度需要 ORM 的那一天再来考虑也不迟。

八、完整的 DB 管理模块

import sqlite3
from flask import g as _g
import os

STATE_DB = os.path.expanduser("~/.hermes/state.db")
MEMORY_DB = os.path.expanduser("~/.agent-memory/memory.db")

def _get_state_db():
    """获取 state.db 连接，请求内复用。不存在则创建。"""
    if "_state_db" not in _g:
        _g._state_db = sqlite3.connect(STATE_DB)
        _g._state_db.row_factory = sqlite3.Row
    return _g._state_db

def _close_state_db():
    """关闭并清理 state.db 连接，从 g 中移除。"""
    db = _g.pop("_state_db", None)
    if db is not None:
        db.close()

def _get_memory_db():
    """获取 memory.db 连接，请求内复用。"""
    if "_mem_db" not in _g:
        _g._mem_db = sqlite3.connect(MEMORY_DB)
    return _g._mem_db

def _close_memory_db():
    """关闭并清理 memory.db 连接。"""
    db = _g.pop("_mem_db", None)
    if db is not None:
        db.close()

九、关键教训

不要用 teardown_appcontext 管理多次连接：它假设一个请求一个连接，如果你的路由内部有"查、关、再查"循环，它会过早关闭或重复关闭
g.pop() 比 g.db = None 更安全：pop 确保 key 真正被移除，下次 _get_state_db() 的 in 检查不会误判
每个连接用完后立即 _close_xxx_db()：不要在路由函数结尾统一关，中间步骤抛异常会导致连接残留
row_factory = sqlite3.Row 让查询结果可以用 row["column_name"] 访问，比 row[0] 可读性高得多

SQLite 加 Flask 撑一个这种规模的站点绰绰有余，不用 ORM，只要把连接的开和关管清楚。等哪天查询复杂到真需要 ORM 了，再上也不迟。