Как копировать данные с одной БД в другую в PostgreSQL?

Копирование данных между базами PostgreSQL

Передача данных между PostgreSQL базами требуется при миграции, резервном копировании или синхронизации систем. Рассмотрю несколько подходов от простых к сложным.

Способ 1: pg_dump + psql (классический)

Самый распространённый и надёжный способ:

# Экспорт всей БД из source_db
pg_dump -h source_host -U source_user -d source_db > dump.sql

# Импорт в target_db
psql -h target_host -U target_user -d target_db < dump.sql

Для больших БД (с компрессией):

# Экспорт в binary формате (быстрее и меньше размер)
pg_dump -h source_host -U source_user -d source_db -Fc > dump.bin

# Импорт
pg_restore -h target_host -U target_user -d target_db dump.bin

Только определённые таблицы:

# Экспорт таблиц users и orders
pg_dump -h source_host -U source_user -d source_db -t users -t orders > dump.sql

# С расширенными параметрами
pg_dump -h source_host -U source_user -d source_db \
  --table=users \
  --table=orders \
  --table='public.comment*' \
  > dump.sql

Способ 2: Прямое копирование между БД (DBLINK)

Для локальных операций без создания промежуточного файла:

-- На целевой БД создаём расширение dblink
CREATE EXTENSION IF NOT EXISTS dblink;

-- Копируем таблицу users
CREATE TABLE users AS
SELECT * FROM dblink(
    'dbname=source_db host=source_host user=source_user password=source_pass',
    'SELECT id, name, email FROM users'
) AS t(id INT, name TEXT, email TEXT);

Для всех таблиц со схемой:

-- Копируем структуру и данные
CREATE TABLE users AS
SELECT * FROM dblink(
    'host=localhost dbname=old_db user=postgres',
    'SELECT * FROM users'
) AS t(id SERIAL PRIMARY KEY, name TEXT, email TEXT);

Способ 3: Программный подход (Python)

Для контроля над процессом и обработки ошибок:

import psycopg2
from contextlib import contextmanager

@contextmanager
db_connection(connection_string):
    conn = psycopg2.connect(connection_string)
    try:
        yield conn
    finally:
        conn.close()

def copy_tables(source_conn_str, target_conn_str, tables):
    """Копирует таблицы из source в target БД."""
    
    with db_connection(source_conn_str) as source:
        with db_connection(target_conn_str) as target:
            source_cur = source.cursor()
            target_cur = target.cursor()
            
            for table in tables:
                print(f"Копирую таблицу {table}...")
                
                # Получаем данные из source
                source_cur.execute(f"SELECT * FROM {table}")
                rows = source_cur.fetchall()
                
                # Получаем информацию о колонках
                source_cur.execute(f"""
                    SELECT column_name FROM information_schema.columns
                    WHERE table_name = '{table}'
                """)
                columns = [col[0] for col in source_cur.fetchall()]
                
                # Вставляем в target
                if rows:
                    placeholders = ','.join(['%s'] * len(columns))
                    insert_query = f"INSERT INTO {table} ({','.join(columns)}) VALUES ({placeholders})"
                    target_cur.executemany(insert_query, rows)
                    target.commit()
                    print(f"  ✓ Вставлено {len(rows)} строк")
                else:
                    print(f"  ✓ Таблица пуста")

# Использование
source = "postgresql://user:pass@source_host/source_db"
target = "postgresql://user:pass@target_host/target_db"
tables = ['users', 'orders', 'comments']

copy_tables(source, target, tables)

Способ 4: Использование COPY для высокой скорости

Для огромных таблиц (миллионы строк):

# Экспорт в CSV
psql -h source_host -U source_user -d source_db -c \
  "COPY users TO STDOUT WITH CSV HEADER" > users.csv

# Импорт из CSV
psql -h target_host -U target_user -d target_db -c \
  "COPY users FROM STDIN WITH CSV HEADER" < users.csv

Python версия:

import csv
import psycopg2

def copy_table_fast(source_conn_str, target_conn_str, table_name):
    """Быстрое копирование через CSV."""
    
    # Экспорт
    source_conn = psycopg2.connect(source_conn_str)
    source_cur = source_conn.cursor()
    
    with open(f"{table_name}.csv", "w") as f:
        source_cur.copy_to(f, table_name, sep=",")
    source_conn.close()
    
    # Импорт
    target_conn = psycopg2.connect(target_conn_str)
    target_cur = target_conn.cursor()
    
    with open(f"{table_name}.csv", "r") as f:
        target_cur.copy_from(f, table_name, sep=",")
    target_conn.commit()
    target_conn.close()
    
    print(f"✓ Таблица {table_name} успешно скопирована")

copy_table_fast(source, target, "users")

Способ 5: WITH foreign data wrapper (FDW)

Для постоянного моста между БД:

-- На целевой БД
CREATE EXTENSION IF NOT EXISTS postgres_fdw;

-- Создаём сервер-источник
CREATE SERVER source_server
    FOREIGN DATA WRAPPER postgres_fdw
    OPTIONS (host 'source_host', dbname 'source_db', port '5432');

-- Маппинг пользователя
CREATE USER MAPPING FOR postgres
    SERVER source_server
    OPTIONS (user 'source_user', password 'source_pass');

-- Импортируем все таблицы из схемы
IMPORT FOREIGN SCHEMA public
    FROM SERVER source_server
    INTO source_schema;

-- Теперь можем копировать как обычный SELECT
CREATE TABLE users AS
SELECT * FROM source_schema.users;

Способ 6: Синхронизация (для текущих данных)

Если нужно синхронизировать изменения:

from datetime import datetime

def sync_incremental(source_conn_str, target_conn_str, table, last_sync):
    """Копирует только новые/изменённые строки."""
    
    with db_connection(source_conn_str) as source:
        source_cur = source.cursor()
        
        # Получаем изменения после last_sync
        source_cur.execute(f"""
            SELECT * FROM {table}
            WHERE updated_at > %s OR created_at > %s
        """, (last_sync, last_sync))
        
        rows = source_cur.fetchall()
    
    if rows:
        with db_connection(target_conn_str) as target:
            target_cur = target.cursor()
            
            # UPSERT вместо INSERT
            for row in rows:
                target_cur.execute(f"""
                    INSERT INTO {table} VALUES {row}
                    ON CONFLICT (id) DO UPDATE SET ...
                """)
            target.commit()
    
    return datetime.now()

Рекомендации по выбору метода

Важные замечания

• Блокировки: pg_dump использует READ COMMITTED, может быть длительная блокировка
• Foreign keys: Отключайте при импорте для скорости: --disable-triggers
• Sequences: Не забудьте обновить: SELECT setval('table_id_seq', MAX(id)) FROM table
• Permissions: Проверьте права доступа на обеих БД
• Транзакции: Оборачивайте операции в транзакции для atomicity

Для production перемещения данных используйте pg_dump с проверкой целостности и backup.