Flask SQLAlchemy中SQLite数据库备份一致性保障方案咨询

阿华AIGC实验室

2026-5-26

针对Flask-SQLAlchemy（SQLite）的一致性数据库备份方案

直接复制app.db确实存在写入时的一致性风险——SQLite在执行写操作时会持有锁，此时复制的文件可能包含部分写入的数据，导致备份损坏或数据不一致。下面是几个官方推荐、实践验证过的优质备份方案，附具体实现：

1. SQLite原生在线备份API（首推方案）

SQLite本身提供了backup()接口，支持不停止应用、不阻塞正常写入的情况下创建一致性备份。Flask-SQLAlchemy可以直接调用底层的SQLite连接来实现这个功能：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import sqlite3

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
db = SQLAlchemy(app)

def backup_database(backup_path='app_backup.db'):
    # 获取Flask-SQLAlchemy底层的SQLite原生连接
    raw_conn = db.engine.raw_connection()
    try:
        # 创建备份文件的连接
        backup_conn = sqlite3.connect(backup_path)
        with backup_conn:
            # 执行在线备份：自动处理锁和数据一致性
            raw_conn.backup(backup_conn)
        print(f"备份成功，文件已保存至 {backup_path}")
    except Exception as e:
        print(f"备份失败：{str(e)}")
    finally:
        # 关闭连接释放资源
        backup_conn.close()
        raw_conn.close()

# 可以通过路由手动触发备份，或者集成到定时任务
@app.route('/trigger-backup')
def trigger_backup():
    backup_database()
    return "数据库备份已完成"

if __name__ == '__main__':
    app.run()

这个方案的优势在于：备份过程中写操作只会被短暂延迟（不会完全阻塞），最终得到的备份文件是完全一致的，是SQLite官方推荐的在线备份方式。

2. 基于只读事务的冷备份（适合低写入场景）

如果你的应用写入频率较低，可以通过开启只读事务锁定数据库状态，再复制文件，确保备份的一致性：

import shutil
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
db = SQLAlchemy(app)

def cold_backup(backup_path='app_backup.db'):
    # 开启只读事务，阻止所有写操作直到事务结束
    with db.engine.connect() as conn:
        conn.execute(db.text("BEGIN READ ONLY"))
        # 复制数据库文件，此时文件状态完全一致
        shutil.copy2('app.db', backup_path)
        # 回滚事务释放锁，恢复正常写入
        conn.rollback()
    print(f"冷备份成功，文件已保存至 {backup_path}")

⚠️ 注意：这个方法会短暂阻塞所有写操作，适合夜间低峰期执行，或者写入量很小的应用。

3. 定时自动化备份（结合任务调度）

为了避免手动操作的遗忘，可以结合任务调度工具（比如APScheduler）实现定时自动备份，同时给备份文件加上时间戳方便管理：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from apscheduler.schedulers.background import BackgroundScheduler
import sqlite3
import datetime

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
db = SQLAlchemy(app)

def auto_backup():
    # 生成带时间戳的备份文件名，避免覆盖旧备份
    timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
    backup_path = f"app_backup_{timestamp}.db"
    
    raw_conn = db.engine.raw_connection()
    try:
        backup_conn = sqlite3.connect(backup_path)
        with backup_conn:
            raw_conn.backup(backup_conn)
        print(f"定时备份成功：{backup_path}")
    except Exception as e:
        print(f"定时备份失败：{str(e)}")
    finally:
        backup_conn.close()
        raw_conn.close()

# 初始化后台调度器，设置每天凌晨2点执行备份
scheduler = BackgroundScheduler()
scheduler.add_job(auto_backup, 'cron', hour=2)
scheduler.start()

@app.route('/')
def index():
    return "应用运行中，自动备份任务已启动"

if __name__ == '__main__':
    try:
        app.run()
    finally:
        # 应用停止时关闭调度器
        scheduler.shutdown()