如何用Python 3.11及FDB库实现Firebird数据库跨服务器表迁移？

下面提供两种基于Python 3.11 + FDB库的全量迁移方案，覆盖不同场景需求：

方案一：全量DDL+数据导出导入（适合中小数据库） #

核心思路是从源库提取所有表的结构（DDL）和数据，再在目标库执行创建表+插入数据的操作。

步骤说明 #

1. 连接源库与目标库
   先初始化两个FDB连接，确保目标库已提前创建空数据库（可通过Firebird工具createdb完成）。

import fdb

# 源数据库连接参数
src_conn_params = {
    'host': '源服务器IP',
    'database': '/path/to/source.fdb',
    'user': 'sysdba',
    'password': 'your_password',
    'charset': 'UTF8'  # 需与源库字符集一致
}

# 目标数据库连接参数
dest_conn_params = {
    'host': '目标服务器IP',
    'database': '/path/to/destination.fdb',
    'user': 'sysdba',
    'password': 'your_password',
    'charset': 'UTF8'
}

src_conn = fdb.connect(**src_conn_params)
dest_conn = fdb.connect(**dest_conn_params)

2. 提取源库所有表名
   通过查询Firebird系统表获取用户自定义表（排除系统表）：

def get_user_tables(conn):
    cursor = conn.cursor()
    # 查询用户表（过滤系统表所在的RDB$ schema）
    cursor.execute("""
        SELECT RDB$RELATION_NAME 
        FROM RDB$RELATIONS 
        WHERE RDB$SYSTEM_FLAG = 0 
        AND RDB$VIEW_BLR IS NULL 
        ORDER BY RDB$RELATION_NAME
    """)
    # 转换为大写（Firebird默认表名大写）
    return [row[0].strip() for row in cursor.fetchall()]

tables = get_user_tables(src_conn)

3. 导出表结构（DDL）并在目标库创建
   利用Firebird内置函数生成DDL，再在目标库执行：

def get_table_ddl(conn, table_name):
    cursor = conn.cursor()
    # 调用Firebird内置函数生成DDL
    cursor.execute(f"SELECT RDB$GET_RDB$RELATION_DDL('{table_name}') FROM RDB$DATABASE")
    ddl = cursor.fetchone()[0].strip()
    # 替换源库相关的存储路径（如果有硬编码）
    ddl = ddl.replace("'源库存储路径'", "'目标库存储路径'")
    return ddl

# 在目标库创建所有表
dest_cursor = dest_conn.cursor()
for table in tables:
    ddl = get_table_ddl(src_conn, table)
    dest_cursor.execute(ddl)
dest_conn.commit()

4. 批量导出并插入数据
   对每个表执行SELECT查询，用executemany批量插入目标库，减少IO开销：

src_cursor = src_conn.cursor()
for table in tables:
    # 获取表所有字段名
    src_cursor.execute(f"SELECT RDB$FIELD_NAME FROM RDB$RELATION_FIELDS WHERE RDB$RELATION_NAME = '{table}' ORDER BY RDB$FIELD_POSITION")
    fields = [row[0].strip() for row in src_cursor.fetchall()]
    field_str = ', '.join(fields)

    # 查询源表所有数据
    src_cursor.execute(f"SELECT {field_str} FROM {table}")
    data = src_cursor.fetchall()

    if data:
        # 生成批量插入语句
        placeholders = ', '.join(['?' for _ in fields])
        insert_sql = f"INSERT INTO {table} ({field_str}) VALUES ({placeholders})"
        dest_cursor.executemany(insert_sql, data)
        dest_conn.commit()

方案二：逐表流式迁移（适合大数据库，避免内存溢出） #

如果数据库数据量较大，一次性读取全表数据会占用过多内存，可采用分批次查询+插入的方式：

def migrate_table_batch(src_conn, dest_conn, table_name, batch_size=1000):
    src_cursor = src_conn.cursor()
    dest_cursor = dest_conn.cursor()

    # 获取字段名
    src_cursor.execute(f"SELECT RDB$FIELD_NAME FROM RDB$RELATION_FIELDS WHERE RDB$RELATION_NAME = '{table}' ORDER BY RDB$FIELD_POSITION")
    fields = [row[0].strip() for row in src_cursor.fetchall()]
    field_str = ', '.join(fields)
    placeholders = ', '.join(['?' for _ in fields])
    insert_sql = f"INSERT INTO {table_name} ({field_str}) VALUES ({placeholders})"

    # 分批次查询数据
    offset = 0
    while True:
        src_cursor.execute(f"SELECT {field_str} FROM {table_name} ROWS {offset} TO {offset + batch_size - 1}")
        batch_data = src_cursor.fetchall()
        if not batch_data:
            break
        dest_cursor.executemany(insert_sql, batch_data)
        dest_conn.commit()
        offset += batch_size

# 逐表迁移
for table in tables:
    migrate_table_batch(src_conn, dest_conn, table)

关键注意事项 #

• 字符集一致性：源库与目标库的字符集必须一致，否则会出现乱码问题。
• 事务处理：每次批量插入后提交事务，避免长时间占用连接和锁资源。
• 索引与约束：方案一的DDL会包含索引、主键等约束，若需单独调整，可在创建表后再修改。
• BLOB字段处理：FDB会自动处理BLOB类型数据，无需额外转换，但需确保目标库支持对应BLOB子类型。
• 权限验证：确保连接用户拥有源库的SELECT权限和目标库的CREATE、INSERT权限。
• 前置准备：目标库需提前创建，可通过Firebird命令行工具执行：createdb -user sysdba -password your_password localhost:/path/to/destination.fdb

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