Azure环境下PostgreSQL生产库每日同步至测试库方案问询

我有两台Azure PostgreSQL服务器，分别部署生产库（prod）和测试库（test），需要每日夜间将生产库备份同步至测试库，确保次日两者完全一致。由于数据表包含约束与主键，仅复制数据无法满足需求，pandas的df.to_sql无法同步架构。原本计划通过Azure Functions的Python脚本每日执行同步，尝试过SQLAlchemy但元数据同步存在严重问题，转而使用PostgreSQL的pg_dump和pg_restore命令通过subprocess实现，当前代码可成功生成备份文件，但无法恢复至测试库。需解决以下问题：

1. 若pg_dump/pg_restore方案可行，Azure Functions中如何指定文件路径？
2. 该方案是否适合在Azure Functions中运行？
3. 若不可行，如何通过SQLAlchemy实现清空测试库数据/元数据并每日从生产库同步？

一、先修复pg_restore的执行错误 #

你的restore_database函数存在参数配置错误，导致恢复失败，修正如下：

def restore_database(location, database, password, username, backup_file):
    cmd = [
        'pg_restore',
        '-Fc',
        '-h', location,
        '-d', database,
        '-U', username,
        '-p', '5432',
        '-W',
        backup_file  # 备份文件直接作为参数传入，无需-f选项
    ]

    try:
        # 传入密码输入，避免交互
        subprocess.run(cmd, check=True, input=password.encode(), capture_output=True, text=True)
        print("Backup restored onto the test server.")
    except subprocess.CalledProcessError as e:
        print("Error occurred while restoring the backup:")
        print(e.stdout)
        print(e.stderr)

错误点说明：

• -f是pg_dump的输出文件参数，pg_restore读取备份文件时直接将文件名作为最后一个参数即可
• 需添加-d指定目标测试库，删除-C（该参数会自动创建数据库，而你的测试库已存在）

二、Azure Functions中的文件路径指定 #

Azure Functions提供几种可用的存储目录，根据需求选择：

• 临时目录：使用tempfile.gettempdir()获取，Linux环境下路径为/tmp，适合存储临时备份文件，函数执行完成后会自动清理，无需手动删除。示例：

  import tempfile
  import os
  backup_file = os.path.join(tempfile.gettempdir(), 'prod_backup.dump')
  

• 函数本地目录：通过os.path.dirname(os.path.abspath(__file__))获取当前脚本所在目录，但该目录空间有限，且函数重启后文件可能丢失，仅适合极小文件。
• Azure Blob Storage：若需要长期保留备份文件或跨实例共享，可将备份文件上传至Blob Storage，可通过Blob挂载或Azure Storage SDK实现。

三、pg_dump/pg_restore方案在Azure Functions的可行性 #

该方案可行，但需注意以下限制：

1. 依赖缺失：Azure Functions默认Python环境未安装PostgreSQL客户端工具（pg_dump/pg_restore），需通过以下方式解决：
   • 构建自定义Docker镜像，包含PostgreSQL客户端和函数代码
   • 使用自定义部署包，将客户端工具打包后部署
2. 执行时间限制：消费计划下函数最长执行时间为10分钟，若数据库较大导致备份+恢复超时，需改用高级计划或专用计划（无时间限制）
3. 权限配置：
   • 生产库和测试库的Azure PostgreSQL服务器需允许Azure Functions的出站IP访问
   • 生产库用户需拥有pg_dump权限，测试库用户需拥有超级用户权限（需删除所有表、恢复架构和数据）

四、SQLAlchemy实现同步方案 #

若不想使用pg_dump/pg_restore，可通过SQLAlchemy实现元数据+数据的完整同步：

1. 清空测试库（含约束处理） #

直接使用metadata.drop_all()可能因外键约束失败，需先禁用外键检查：

from sqlalchemy import text

def clear_test_database(engine):
    with engine.connect() as conn:
        # 禁用外键约束，避免删表失败
        conn.execute(text("SET session_replication_role = 'replica';"))
        # 获取public schema下的所有表
        tables = conn.execute(text("SELECT tablename FROM pg_tables WHERE schemaname = 'public';")).fetchall()
        for (tablename,) in tables:
            conn.execute(text(f"DROP TABLE IF EXISTS {tablename} CASCADE;"))
        # 恢复外键约束
        conn.execute(text("SET session_replication_role = 'origin';"))
        conn.commit()

2. 同步生产库架构到测试库 #

from sqlalchemy import MetaData

def sync_schema(prod_engine, test_engine):
    # 读取生产库的完整元数据（表、约束、索引等）
    prod_metadata = MetaData()
    prod_metadata.reflect(bind=prod_engine)
    # 在测试库创建所有架构对象
    prod_metadata.create_all(bind=test_engine)

3. 批量同步所有表数据 #

from sqlalchemy import select

def sync_data(prod_engine, test_engine):
    prod_metadata = MetaData()
    prod_metadata.reflect(bind=prod_engine)
    
    with prod_engine.connect() as prod_conn, test_engine.connect() as test_conn:
        # 按外键依赖顺序同步表，避免插入失败
        for table in prod_metadata.sorted_tables:
            # 读取生产库表的所有数据
            data = prod_conn.execute(select(table)).fetchall()
            if data:
                # 批量插入测试库
                test_conn.execute(table.insert(), data)
        test_conn.commit()

4. 完整同步流程 #

from sqlalchemy import create_engine

# 初始化生产库和测试库引擎
prod_engine = create_engine("postgresql://prod_user:prod_pass@prod_host:5432/prod_db")
test_engine = create_engine("postgresql://test_user:test_pass@test_host:5432/test_db")

# 执行同步
clear_test_database(test_engine)
sync_schema(prod_engine, test_engine)
sync_data(prod_engine, test_engine)

注意：该方案适合中小型数据库，若数据量过大，需添加分页读取逻辑避免内存溢出，此时pg_dump/pg_restore的效率会更高。

内容的提问来源于stack exchange，提问作者Marcelo ARK