Python数据API数据备份怎么策略

本文目录导读：

1. 第一阶段：策略设计（先定规则再写代码）
2. 第二阶段：具体实现（Python脚本）
3. 第三阶段：云存储与异地容灾（关键！）
4. 第四阶段：恢复测试（最重要的一步）
5. 给你的API建议

针对Python数据API的备份策略，核心目标是在数据丢失（误删、损坏、被黑）或系统故障时,能够快速恢复服务。

下面是一套从策略设计到代码实现的完整方案，涵盖最常用的三种场景：JSON/CSV文件、SQLite/MySQL数据库、Redis缓存。

第一阶段：策略设计（先定规则再写代码）

第二阶段：具体实现（Python脚本）

以下代码覆盖文件型数据、SQLite、MySQL、Redis四种常见API后端。

通用工具函数（备份文件、验证、清理旧备份）

import os
import shutil
import hashlib
import datetime
import schedule
import time
BACKUP_ROOT = "/data/backups"  # 备份根目录
RETENTION_DAYS = 7            # 保留天数
def calculate_md5(file_path):
    """计算文件MD5用于校验"""
    hash_md5 = hashlib.md5()
    with open(file_path, "rb") as f:
        for chunk in iter(lambda: f.read(4096), b""):
            hash_md5.update(chunk)
    return hash_md5.hexdigest()
def clean_old_backups(backup_type_dir):
    """删除指定目录下超过保留天数的备份"""
    now = time.time()
    for f in os.listdir(backup_type_dir):
        full_path = os.path.join(backup_type_dir, f)
        if os.path.isfile(full_path):
            # 从文件名解析日期，假设格式为: data_backup_20231001_120000.tar.gz
            parts = f.split('_')
            if len(parts) >= 3:
                try:
                    file_time = datetime.datetime.strptime(parts[-2] + parts[-1].split('.')[0], "%Y%m%d%H%M%S")
                    if (datetime.datetime.now() - file_time).days > RETENTION_DAYS:
                        os.remove(full_path)
                        print(f"【清理】删除过期备份: {f}")
                except ValueError:
                    pass
def verify_backup(backup_path):
    """验证备份文件完整性（可扩展为解压校验）"""
    return os.path.exists(backup_path) and os.path.getsize(backup_path) > 0

备份JSON/CSV文件型数据

def backup_json_data(api_data_path, backup_category="json"):
    """备份API使用的JSON/CSV数据文件"""
    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_dir = os.path.join(BACKUP_ROOT, backup_category)
    os.makedirs(backup_dir, exist_ok=True)
    backup_filename = f"{backup_category}_backup_{timestamp}.tar.gz"
    backup_fullpath = os.path.join(backup_dir, backup_filename)
    # 压缩备份（比单纯复制节省空间）
    shutil.make_archive(backup_fullpath.replace('.tar.gz', ''), 'gztar', api_data_path)
    # 校验
    md5 = calculate_md5(backup_fullpath)
    with open(backup_fullpath + ".md5", "w") as f:
        f.write(md5)
    # 清理旧备份
    clean_old_backups(backup_dir)
    # 同步到云存储（示例：阿里云OSS/S3）
    # upload_to_oss(backup_fullpath)
    print(f"✅ 文件数据备份成功: {backup_fullpath} (MD5: {md5[:8]}...)")
    return backup_fullpath
# 使用： backup_json_data("/data/api/data/")

备份SQLite数据库（API常用轻量级库）

SQLite是文件型数据库，直接复制可能会损坏（写入中），需要使用 sqlite3 模块的逻辑备份。

import sqlite3
def backup_sqlite(db_path, backup_category="sqlite"):
    """安全的SQLite在线备份（VACUUM INTO 或 .backup）"""
    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_dir = os.path.join(BACKUP_ROOT, backup_category)
    os.makedirs(backup_dir, exist_ok=True)
    backup_filename = f"{backup_category}_backup_{timestamp}.db"
    backup_fullpath = os.path.join(backup_dir, backup_filename)
    # 方法1：使用Python sqlite3 的备份API（推荐，支持热备份）
    source_conn = sqlite3.connect(db_path)
    backup_conn = sqlite3.connect(backup_fullpath)
    source_conn.backup(backup_conn, pages=1024)  # 逐页复制，不锁原库
    source_conn.close()
    backup_conn.close()
    # 校验
    md5 = calculate_md5(backup_fullpath)
    with open(backup_fullpath + ".md5", "w") as f:
        f.write(md5)
    clean_old_backups(backup_dir)
    print(f"✅ SQLite备份成功: {backup_fullpath}")
    return backup_fullpath
# 使用： backup_sqlite("/data/api/mydatabase.db")

备份MySQL数据库（使用mysqldump）

import subprocess
def backup_mysql(host='localhost', user='root', password='yourpass', db_name='api_db', backup_category="mysql"):
    """使用mysqldump备份MySQL（生产环境建议用 xtrabackup 做物理备份）"""
    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_dir = os.path.join(BACKUP_ROOT, backup_category)
    os.makedirs(backup_dir, exist_ok=True)
    backup_filename = f"{backup_category}_backup_{timestamp}.sql.gz"
    backup_fullpath = os.path.join(backup_dir, backup_filename)
    # 压缩导出
    dump_cmd = f"mysqldump -h{host} -u{user} -p{password} --single-transaction --routines --triggers {db_name} | gzip > {backup_fullpath}"
    try:
        subprocess.run(dump_cmd, shell=True, check=True, timeout=3600)
        md5 = calculate_md5(backup_fullpath)
        with open(backup_fullpath + ".md5", "w") as f:
            f.write(md5)
        clean_old_backups(backup_dir)
        print(f"✅ MySQL备份成功: {backup_fullpath}")
    except subprocess.CalledProcessError as e:
        print(f"❌ MySQL备份失败: {e}")
        return None
    return backup_fullpath

备份Redis缓存数据（RDB快照）

import redis
def backup_redis(redis_host='localhost', redis_port=6379, backup_category="redis"):
    """对Redis执行BGSAVE并复制RDB文件"""
    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_dir = os.path.join(BACKUP_ROOT, backup_category)
    os.makedirs(backup_dir, exist_ok=True)
    client = redis.Redis(host=redis_host, port=redis_port)
    # 触发BGSAVE（异步，不阻塞服务）
    client.bgsave()
    print("⏳ Redis BGSAVE 已触发，等待完成...")
    # 等待BGSAVE完成（最多30秒）
    for _ in range(30):
        info = client.info("persistence")
        if info.get("rdb_bgsave_in_progress") == 0:
            break
        time.sleep(1)
    # 获取RDB文件路径（通常在 /var/lib/redis/dump.rdb）
    rdb_path = client.config_get("dir")["dir"] + "/dump.rdb"
    if os.path.exists(rdb_path):
        backup_filename = f"{backup_category}_backup_{timestamp}.rdb"
        backup_fullpath = os.path.join(backup_dir, backup_filename)
        shutil.copy2(rdb_path, backup_fullpath)
        md5 = calculate_md5(backup_fullpath)
        with open(backup_fullpath + ".md5", "w") as f:
            f.write(md5)
        clean_old_backups(backup_dir)
        print(f"✅ Redis备份成功: {backup_fullpath}")
        return backup_fullpath
    else:
        print("❌ Redis RDB文件未找到")
        return None

统一调度器（定时任务）

def daily_full_backup():
    """每日凌晨2点执行全量备份"""
    print(f"开始每日全量备份: {datetime.datetime.now()}")
    backup_json_data("/data/api/data/")
    backup_sqlite("/data/api/mydatabase.db")
    backup_mysql()
    backup_redis()
    print("全量备份完成")
def hourly_incremental_backup():
    """每小时增量备份（仅针对主要数据文件）"""
    print(f"开始增量备份: {datetime.datetime.now()}")
    # 增量备份方案（示例：仅备份变化文件）
    # rsync --link-dest=/data/backups/json/previous_full /data/api/data/ /data/backups/json/incremental/
    print("增量备份完成")
# 定时任务配置
schedule.every().day.at("02:00").do(daily_full_backup)
schedule.every(1).hour.do(hourly_incremental_backup)
if __name__ == "__main__":
    print("备份守护进程启动中...")
    while True:
        schedule.run_pending()
        time.sleep(60)  # 每分钟检查一次

第三阶段：云存储与异地容灾（关键！）

仅存在本地硬盘的备份是不安全的（硬盘损坏、勒索病毒）。

def sync_to_cloud(local_backup_path, cloud_path="backups/api/"):
    """同步到阿里云OSS / AWS S3（需安装boto3）"""
    import boto3
    s3 = boto3.client('s3', 
                      aws_access_key_id='your_key',
                      aws_secret_access_key='your_secret')
    s3.upload_file(local_backup_path, 'your-bucket', 
                   cloud_path + os.path.basename(local_backup_path))
    print(f"☁️ 已同步到云存储: {cloud_path}")

或者使用 rsync 到另一台服务器（增量、带压缩）：

rsync -avz --delete /data/backups/ user@remote-server:/data/backups/

第四阶段：恢复测试（最重要的一步）

创建 restore_test.py,每月自动跑一次：

def test_restore_json(backup_file, restore_path="/tmp/test_restore"):
    """解压备份到临时目录，校验文件完整性"""
    shutil.rmtree(restore_path, ignore_errors=True)
    shutil.unpack_archive(backup_file, restore_path)
    # 校验MD5
    # 模拟API启动，测试读取
    print(f"✅ JSON恢复测试通过: {backup_file}")
def test_restore_sqlite(backup_db):
    """恢复SQLite并执行简单查询"""
    conn = sqlite3.connect(backup_db)
    cursor = conn.cursor()
    cursor.execute("SELECT COUNT(*) FROM sqlite_master WHERE type='table'")
    tables = cursor.fetchone()[0]
    print(f"✅ SQLite恢复测试: 包含 {tables} 个表")
    conn.close()

给你的API建议

最后提醒：不要把备份脚本和业务代码混在一起！单独部署一个 Backup Service（可以是一个独立的Python进程 + crontab），这样即使API挂了,备份依然能运行。