"""数据库备份执行器 — 使用 mysqldump/pg_dump 执行真实备份。 支持: - MySQL: mysqldump + gzip 压缩 - PostgreSQL: pg_dump custom format 备份文件存储在 /app/backups 目录。 """ import os import asyncio from datetime import datetime from typing import Optional, Tuple from app.models.instance import Instance from app.models.backup import Backup from app.core.crypto import decrypt_password # 备份存储目录 BACKUP_DIR = os.environ.get("BACKUP_DIR", "/app/backups") class BackupExecutor: """数据库备份执行器。""" async def execute(self, backup: Backup, instance: Instance) -> Tuple[str, int]: """执行备份,返回 (文件路径, 文件大小字节)。 Raises: RuntimeError: 备份工具未安装或备份失败 """ os.makedirs(BACKUP_DIR, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") password = decrypt_password(instance.password) if instance.db_type == "mysql": filepath = await self._backup_mysql( instance, backup.database, password, timestamp ) elif instance.db_type in ("postgres", "postgresql"): filepath = await self._backup_postgres( instance, backup.database, password, timestamp ) else: raise ValueError(f"不支持的备份类型: {instance.db_type}") size_bytes = os.path.getsize(filepath) if os.path.exists(filepath) else 0 return filepath, size_bytes async def _backup_mysql(self, instance: Instance, database: str, password: str, timestamp: str) -> str: """使用 mysqldump 备份 MySQL 数据库。 输出通过 gzip 压缩存储。 """ filename = f"{database}_mysql_{timestamp}.sql.gz" filepath = os.path.join(BACKUP_DIR, filename) # 检查 mysqldump 是否可用 try: proc = await asyncio.create_subprocess_exec( "mysqldump", "--version", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) await proc.communicate() if proc.returncode != 0: raise RuntimeError("mysqldump 未安装") except FileNotFoundError: raise RuntimeError( "mysqldump 未安装,请安装 MySQL 客户端工具:\\n" " apt-get install mysql-client 或\\n" " yum install mysql" ) cmd = [ "mysqldump", "-h", instance.host, "-P", str(instance.port), "-u", instance.username, f"-p{password}", "--single-transaction", "--routines", "--triggers", "--events", "--hex-blob", "--skip-lock-tables", database, ] # 创建 mysqldump 子进程,stdout 通过 gzip 压缩 dump_proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) # 创建 gzip 子进程 gzip_proc = await asyncio.create_subprocess_exec( "gzip", "-c", stdin=dump_proc.stdout, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) # 关闭 dump_proc 的 stdout 在本进程的引用 if dump_proc.stdout: dump_proc.stdout.close() # 读取 gzip 输出写入文件 try: with open(filepath, "wb") as f: while True: chunk = await gzip_proc.stdout.read(65536) if gzip_proc.stdout else b"" if not chunk: break f.write(chunk) # 等待进程完成 dump_stderr = await dump_proc.stderr.read() if dump_proc.stderr else b"" gzip_stderr = await gzip_proc.stderr.read() if gzip_proc.stderr else b"" await dump_proc.wait() await gzip_proc.wait() if dump_proc.returncode != 0: error_msg = dump_stderr.decode("utf-8", errors="ignore")[-500:] raise RuntimeError(f"mysqldump 执行失败: {error_msg}") if gzip_proc.returncode != 0: error_msg = gzip_stderr.decode("utf-8", errors="ignore") raise RuntimeError(f"gzip 压缩失败: {error_msg}") except Exception: # 清理失败的文件 if os.path.exists(filepath): os.remove(filepath) raise return filepath async def _backup_postgres(self, instance: Instance, database: str, password: str, timestamp: str) -> str: """使用 pg_dump 备份 PostgreSQL 数据库。 使用 custom format(已内置压缩)。 """ filename = f"{database}_postgres_{timestamp}.dump" filepath = os.path.join(BACKUP_DIR, filename) # 检查 pg_dump 是否可用 try: proc = await asyncio.create_subprocess_exec( "pg_dump", "--version", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) await proc.communicate() if proc.returncode != 0: raise RuntimeError("pg_dump 未安装") except FileNotFoundError: raise RuntimeError( "pg_dump 未安装,请安装 PostgreSQL 客户端工具:\\n" " apt-get install postgresql-client 或\\n" " yum install postgresql" ) env = os.environ.copy() env["PGPASSWORD"] = password cmd = [ "pg_dump", "-h", instance.host, "-p", str(instance.port), "-U", instance.username, "-d", database, "-F", "c", # custom format (compressed) "-Z", "5", # compression level 5 "--no-owner", # don't add ownership commands "--no-privileges", # don't add privilege commands "-f", filepath, ] proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=env, ) stdout, stderr = await proc.communicate() if proc.returncode != 0: error_msg = stderr.decode("utf-8", errors="ignore")[-500:] # 清理失败的文件 if os.path.exists(filepath): os.remove(filepath) raise RuntimeError(f"pg_dump 执行失败: {error_msg}") return filepath async def restore_mysql(self, instance: Instance, database: str, filepath: str) -> str: """从备份恢复 MySQL 数据库。 Args: instance: 目标实例 database: 目标数据库 filepath: 备份文件路径(.sql.gz) """ password = decrypt_password(instance.password) if filepath.endswith(".gz"): # gunzip 解压后通过 mysql 导入 gunzip_proc = await asyncio.create_subprocess_exec( "gunzip", "-c", filepath, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) mysql_cmd = [ "mysql", "-h", instance.host, "-P", str(instance.port), "-u", instance.username, f"-p{password}", database, ] mysql_proc = await asyncio.create_subprocess_exec( *mysql_cmd, stdin=gunzip_proc.stdout, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) if gunzip_proc.stdout: gunzip_proc.stdout.close() await mysql_proc.wait() await gunzip_proc.wait() if mysql_proc.returncode != 0: raise RuntimeError("MySQL 恢复失败") else: # 直接导入 .sql 文件 with open(filepath, "rb") as f: mysql_cmd = [ "mysql", "-h", instance.host, "-P", str(instance.port), "-u", instance.username, f"-p{password}", database, ] proc = await asyncio.create_subprocess_exec( *mysql_cmd, stdin=f, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) _, stderr = await proc.communicate() if proc.returncode != 0: raise RuntimeError(f"MySQL 恢复失败: {stderr.decode()[-500:]}") return f"数据库 {database} 恢复成功" async def restore_postgres(self, instance: Instance, database: str, filepath: str) -> str: """从备份恢复 PostgreSQL 数据库。 Args: instance: 目标实例 database: 目标数据库 filepath: 备份文件路径(.dump) """ password = decrypt_password(instance.password) env = os.environ.copy() env["PGPASSWORD"] = password cmd = [ "pg_restore", "-h", instance.host, "-p", str(instance.port), "-U", instance.username, "-d", database, "--no-owner", "--no-privileges", "-c", # clean (drop) database objects before recreating filepath, ] proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=env, ) _, stderr = await proc.communicate() # pg_restore 即使有一些警告也可能成功 if proc.returncode not in (0, 1): # 1 = warnings error_msg = stderr.decode("utf-8", errors="ignore")[-500:] raise RuntimeError(f"pg_restore 失败: {error_msg}") return f"数据库 {database} 恢复成功" @staticmethod def validate_backup_file(filepath: str) -> bool: """验证备份文件是否有效。 - MySQL: 检查 gzip 文件完整性 - PostgreSQL: 检查 custom format 文件 """ if not os.path.exists(filepath): return False if filepath.endswith(".gz"): # 使用 gzip -t 测试 import subprocess try: result = subprocess.run( ["gzip", "-t", filepath], capture_output=True, timeout=30, ) return result.returncode == 0 except (subprocess.TimeoutExpired, FileNotFoundError): # 简单检查文件大小和魔数 with open(filepath, "rb") as f: magic = f.read(2) return magic == b"\\x1f\\x8b" # gzip magic number elif filepath.endswith(".dump"): # PostgreSQL custom format: 检查文件大小和魔数 if os.path.getsize(filepath) < 100: return False with open(filepath, "rb") as f: header = f.read(5) # PostgreSQL custom format starts with "PGDMP" return header == b"PGDMP" return os.path.getsize(filepath) > 0 @staticmethod def get_backup_size(filepath: str) -> int: """获取备份文件大小(字节)。""" if os.path.exists(filepath): return os.path.getsize(filepath) return 0