如何在Python中备份SQLite数据库？Telegram Bot数据备份异常求助

问题根源分析 #

1. 数据库连接管理错误：你在/start命令处理函数中每次执行后都关闭了全局数据库连接，后续/stat、/backup命令使用已关闭的游标时，会触发隐式重新连接，导致数据未正确持久化甚至丢失。
2. 备份逻辑无效：当前备份函数连接的是同一个users.db文件，相当于将数据库备份到自身，完全起不到备份作用。
3. 延迟写入导致文件旧数据：SQLite默认采用延迟写入机制，即使执行commit，数据可能仍停留在内存缓存中，直接复制文件会得到未更新的旧数据。

修复步骤与优化代码 #

1. 正确管理数据库连接

• 全局初始化数据库连接，仅在Bot停止时关闭，避免单个命令处理函数中随意关闭连接。
• 添加check_same_thread=False适配aiogram异步框架的多线程场景。

2. 修复备份逻辑

• 备份到独立的目标文件，备份前强制将内存缓存刷入磁盘，确保备份文件是最新数据。

3. 优化后的完整代码

import sqlite3
from datetime import datetime
from aiogram import Bot, Dispatcher
from aiogram.types import Message
from aiogram.utils import executor

API_TOKEN = '你的Bot令牌'
bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)

# 全局数据库连接，启动时初始化，停止时关闭
connection = sqlite3.connect('users.db', check_same_thread=False)
cursor = connection.cursor()
# 确保用户表存在
cursor.execute(
    'CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY AUTOINCREMENT, chat_id INTEGER, name VARCHAR(255))')
connection.commit()


@dp.message_handler(commands=['start'])
async def send_welcome(message: Message):
    chat_id = message.chat.id
    user_name = message.from_user.first_name
    # 查询用户是否已存在
    cursor.execute('SELECT count(chat_id) FROM users WHERE chat_id = :chat_id', {'chat_id': chat_id})
    is_user = cursor.fetchone()[0]
    if not is_user:
        cursor.execute("INSERT INTO users (chat_id, name) VALUES (?, ?)", (chat_id, user_name))
        connection.commit()
    # 不要在此处关闭连接
    await bot.send_message(
        chat_id,
        f'Hello {user_name}, this bot helps you to download media files from social medias such as *tiktok, instagram, youtube, pinterest*',
        parse_mode='markdownv2'
    )


admins = [679679313]

@dp.message_handler(commands=['stat'])
async def send_stat(message: Message):
    chat_id = message.chat.id
    if chat_id not in admins:
        return
    current_date = datetime.now().strftime("%B %d, %Y %H:%M:%S")
    cursor.execute('SELECT count(*) FROM users')
    user_count = cursor.fetchone()[0]
    await bot.send_message(
        chat_id,
        f"""🤖Bot Statistics
👤 Users : {user_count}
🗓️ {current_date}"""
    )


def backup_database():
    # 备份到独立文件
    backup_conn = sqlite3.connect('users_backup.db')
    # 强制将所有缓存数据写入磁盘
    connection.execute("PRAGMA wal_checkpoint(FULL)")
    connection.commit()
    # 执行备份
    connection.backup(backup_conn)
    backup_conn.close()
    return 'users_backup.db'


@dp.message_handler(commands=['backup'])
async def send_backup(message: Message):
    chat_id = message.chat.id
    if chat_id not in admins:
        return
    backup_file = backup_database()
    await bot.send_document(chat_id, open(backup_file, "rb"))


# Bot停止时关闭数据库连接
async def on_shutdown(dispatcher):
    connection.close()


if __name__ == '__main__':
    executor.start_polling(dp, on_shutdown=on_shutdown)

额外注意事项 #

• 部署持久化：如果使用Docker等容器部署，需将数据库所在目录挂载为持久化卷，例如docker run -v ./bot_data:/app your-bot-image，防止容器重启时数据丢失。
• 高并发场景优化：如果Bot用户量较大，建议使用线程池处理数据库操作，避免check_same_thread=False带来的线程安全问题。
• 自动备份：可集成APScheduler等定时任务库，实现每日/每周自动备份，提升数据安全性。

内容的提问来源于stack exchange，提问作者Jakhongir Ganiev