""" 第 14 章 · 演示 4：用 Python 调度 pg_dump 做定时备份 ====================================================== 场景：用 psycopg + subprocess 实现一个「生产级备份脚本」，特点： 1) 自动按日期建子目录 /backup/2026-04-17/ 2) 用 -Fd -j N 并行 dump 3) 校验 dump 完整性（pg_restore -l） 4) 写入备份元信息到一张审计表 ch14_backup_history 5) 自动清理 N 天前的旧备份 6) 自动 dump 全局对象 (pg_dumpall -g) 依赖： pip install "psycopg[binary]>=3.1" 用法： python 04_dump_with_python.py --dest /tmp/pg_backup --jobs 4 --keep 7 可放进 cron： 0 2 * * * /usr/bin/python /path/to/04_dump_with_python.py --dest /backup --keep 30 """ from __future__ import annotations import argparse import datetime as dt import os import shutil import subprocess import sys from pathlib import Path import psycopg DEFAULT_DSN = "host=127.0.0.1 port=5432 dbname=learn_pg user=postgres" HISTORY_TABLE_DDL = """ CREATE TABLE IF NOT EXISTS ch14_backup_history ( id BIGSERIAL PRIMARY KEY, backup_name TEXT NOT NULL, backup_path TEXT NOT NULL, format TEXT, size_bytes BIGINT, started_at TIMESTAMPTZ NOT NULL, finished_at TIMESTAMPTZ NOT NULL, ok BOOLEAN NOT NULL, error_msg TEXT ); """ def run(cmd: list[str], **kw) -> subprocess.CompletedProcess: """执行命令，stderr/stdout 实时打印；非 0 抛异常。""" print(f" $ {' '.join(cmd)}") return subprocess.run(cmd, check=True, text=True, **kw) def folder_size(path: Path) -> int: if path.is_file(): return path.stat().st_size return sum(p.stat().st_size for p in path.rglob('*') if p.is_file()) def ensure_history_table(dsn: str) -> None: with psycopg.connect(dsn) as c, c.cursor() as cur: cur.execute(HISTORY_TABLE_DDL) c.commit() def record_history(dsn: str, **kw) -> None: sql = """ INSERT INTO ch14_backup_history (backup_name, backup_path, format, size_bytes, started_at, finished_at, ok, error_msg) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) """ with psycopg.connect(dsn) as c, c.cursor() as cur: cur.execute(sql, ( kw["backup_name"], kw["backup_path"], kw["format"], kw["size_bytes"], kw["started_at"], kw["finished_at"], kw["ok"], kw.get("error_msg"), )) c.commit() def backup_once(args: argparse.Namespace) -> Path: today = dt.datetime.now().strftime("%Y-%m-%d_%H%M%S") backup_root = Path(args.dest) backup_root.mkdir(parents=True, exist_ok=True) sub = backup_root / today sub.mkdir(exist_ok=True) print(f"\n=== 开始备份 → {sub} ===") started = dt.datetime.now(dt.timezone.utc) err: str | None = None try: # 1) 全局对象（角色、表空间） run([ "pg_dumpall", "-h", args.host, "-p", str(args.port), "-U", args.user, "--globals-only", "-f", str(sub / "globals.sql"), ]) # 2) 业务库 -Fd 并行 dump run([ "pg_dump", "-h", args.host, "-p", str(args.port), "-U", args.user, "-d", args.dbname, "-Fd", "-j", str(args.jobs), "-Z", "6", "-f", str(sub / "main"), "--verbose", ]) # 3) 校验完整性 proc = subprocess.run( ["pg_restore", "-l", str(sub / "main")], check=True, text=True, capture_output=True, ) n_items = sum(1 for ln in proc.stdout.splitlines() if ln.strip() and not ln.startswith(';')) print(f" ✅ pg_restore -l 校验通过，共 {n_items} 个对象") # 4) 计算大小 size = folder_size(sub) print(f" 📦 备份总大小: {size/1024/1024:.2f} MB") finished = dt.datetime.now(dt.timezone.utc) record_history( args.dsn, backup_name=today, backup_path=str(sub), format="directory", size_bytes=size, started_at=started, finished_at=finished, ok=True, ) print(f" ⏱️ 耗时 {(finished-started).total_seconds():.1f}s") return sub except subprocess.CalledProcessError as e: err = f"{e.cmd} returned {e.returncode}" print(f" ❌ 备份失败: {err}") finished = dt.datetime.now(dt.timezone.utc) record_history( args.dsn, backup_name=today, backup_path=str(sub), format="directory", size_bytes=0, started_at=started, finished_at=finished, ok=False, error_msg=err, ) raise def cleanup_old(dest: Path, keep_days: int) -> None: if keep_days <= 0: return cutoff = dt.datetime.now() - dt.timedelta(days=keep_days) print(f"\n=== 清理 {keep_days} 天前的旧备份（< {cutoff}） ===") for child in dest.iterdir(): if not child.is_dir(): continue try: d = dt.datetime.strptime(child.name.split("_")[0], "%Y-%m-%d") except ValueError: continue if d < cutoff: print(f" 🗑️ 删除 {child}") shutil.rmtree(child, ignore_errors=True) def parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(description="PG 定时备份 (Python 调度 pg_dump)") p.add_argument("--host", default="127.0.0.1") p.add_argument("--port", type=int, default=5432) p.add_argument("--user", default="postgres") p.add_argument("--dbname", default="learn_pg") p.add_argument("--dest", default="/tmp/pg_backup", help="备份根目录") p.add_argument("--jobs", type=int, default=4, help="pg_dump 并行 worker 数") p.add_argument("--keep", type=int, default=7, help="保留天数") p.add_argument("--dsn", default=DEFAULT_DSN, help="psycopg 连接串（用于写 ch14_backup_history）") return p.parse_args() def main() -> int: args = parse_args() ensure_history_table(args.dsn) backup_once(args) cleanup_old(Path(args.dest), args.keep) print("\n=== 最近 5 次备份历史 ===") with psycopg.connect(args.dsn) as c, c.cursor() as cur: cur.execute(""" SELECT id, backup_name, format, pg_size_pretty(size_bytes::bigint), to_char(started_at, 'MM-DD HH24:MI:SS'), round(extract(epoch from (finished_at - started_at))::numeric, 1) || 's', ok FROM ch14_backup_history ORDER BY id DESC LIMIT 5 """) rows = cur.fetchall() print(f" {'ID':<5}{'Name':<22}{'Fmt':<12}{'Size':<10}{'Time':<18}{'Cost':<8}OK") for r in rows: print(f" {r[0]:<5}{r[1]:<22}{r[2] or '':<12}{r[3] or '':<10}{r[4]:<18}{r[5] or '':<8}{r[6]}") return 0 if __name__ == "__main__": sys.exit(main())