Flask独立线程中使用Peewee ORM安全更新数据库的问题咨询

问题核心原因 #

• Peewee的数据库连接默认采用线程本地存储实现，不同线程的连接互相隔离
• 你配置的before_request/after_request钩子仅作用于处理HTTP请求的Flask工作线程，自定义的后台任务线程不受该钩子管控，线程内的数据库连接没有被正确管理，高并发场景下会出现连接泄漏、连接被意外关闭、数据库操作超时等错误。

最简修复方案 #

直接在后台任务函数内手动用db上下文管理器包裹数据库操作即可，Peewee会自动为当前线程创建、回收独立的数据库连接，不会和请求线程的连接产生冲突：

def do_smth(task_id):
    start = time()
    sleep(10)
    # 手动管理当前线程的数据库连接生命周期
    with db:
        Task.update({Task.status: 1, Task.result: f"{start} - {time()}"})\
            .where(Task.id == task_id).execute()

你之前设想的调用专用路由的方案确实不合理，会额外增加HTTP请求开销，还要额外处理路由鉴权、请求异常等问题，完全没有必要。

额外优化建议 #

• 如果使用SQLite作为数据库，它的写锁是全局库级别的，高并发写场景容易出现锁等待超时，可以在初始化数据库时增加超时参数：db = SqliteDatabase("test.db", timeout=30)，超时时间单位为秒，设置为大于你的任务最长耗时即可。
• 若后续并发量持续升高，手动创建线程会导致系统资源占用过高，可以换成固定容量的线程池来执行后台任务，避免资源耗尽：

from concurrent.futures import ThreadPoolExecutor
# 初始化最大容量为10的线程池，可根据业务调整
executor = ThreadPoolExecutor(max_workers=10)

# 提交任务的地方改成
executor.submit(do_smth, task.id)

• 如果后续迁移到MySQL/PostgreSQL这类关系型数据库，可以直接使用Peewee提供的连接池实现，进一步复用连接提升性能。

内容的提问来源于stack exchange，提问作者Olvin Roght