最近遇到一個場景需要從一個postgresql庫同步一張表到另一個postgresql庫中，但又不需要實時同步，就寫了個同步的代碼，本來網(wǎng)上同步的方法早都有了，之所以自己寫一套，是因為postgresql數(shù)據(jù)庫可用的太少了，于是我決定擼起袖子再寫一套。

整個代碼部分就不再過多啰嗦了，因為都是一些基礎，目的只有一個：讓你快速可以使用。如果有同樣的需求，改下配置settings就可以直接用。如果的確有看不懂的地方，請把你的疑惑留在評論區(qū)，如果沒有，那我的目的就達到了。整塊代碼主要用到兩個方法copy_to、copy_from

• copy_to用于把一個表的內(nèi)容復制到一個文件；copy_to中也可以指定查詢，將查詢結(jié)果寫入文件
• copy_from從文件復制數(shù)據(jù)到表中。copy_from中，文件的字段按照順序?qū)懭氲街付兄小?/li>

需要注意的是：
1.數(shù)據(jù)庫用戶必須有文件所在的路徑的寫權(quán)限。
2.表中存在中文時要考慮編碼問題

• 上菜??

import os
import datetime
import logging.config

from settings import log_config,local_data_home

logging.config.dictConfig(log_config)
logger = logging.getLogger(__name__)


def get_conn(sys_code='SOURCE'):
    """
    數(shù)據(jù)庫連接獲取
    :return:
    """
    params = db_param[sys_code]
    host = params['host']
    port = params['port']
    database = params['dbname']
    user = params['user']
    password = params['password']
    db_type = params['DBType'].upper()
    if db_type == "PostgreSQL".upper():
        return psycopg2.connect(database=database, user=user, password=password, host=host, port=port)
    if db_type == "Mysql".upper():
        return pymysql.connect(database=database, user=user, password=password, host=host, port=port)
    elif db_type == "Oracle".upper():
        os.environ['NLS_LANG'] = 'SIMPLIFIED CHINESE_CHINA.UTF8'
        dsn = cx_Oracle.makedsn(host, port, service_name=database)
        conn = cx_Oracle.connect(user, password, dsn=dsn)
        return conn
    elif db_type == 'SQLServer'.upper():
        return pymssql.connect(host=host, user=user, password=password, database=database, charset="utf8")
    elif db_type == 'Mongodb'.upper():
        if len(user) > 0 and len(password) > 0:
            conn_url = 'mongodb://{user}:{password}@{host}:{port}'
        else:
            conn_url = 'mongodb://{host}:{port}'
        return pymongo.MongoClient(conn_url.format(**params))
    else:
        raise Exception("源系統(tǒng)%s數(shù)據(jù)庫連接失敗. " % sys_code)

# column：要被復制的列列表
def get_column(table_name):
    conn = None
    columns = []
    try:
        sql = "SELECT * FROM %s LIMIT 1" % table_name
        conn = get_conn('SOURCE')
        with conn.cursor() as cur:
            cur.execute(sql)
            for d in cur.description:
                columns.append(d.name)
    finally:
        if conn:
            conn.close()
    return columns


def get_file_name(prefix, suffix='txt'):
    """
    返回文件名
    :param prefix:
    :param suffix:
    :return:
    """
    return prefix.lower() + '.' + suffix


# 表名小寫
def get_file_prefix(s_table_name):
    return s_table_name.lower()


def get_local_path(s_table_name):
    """
    本地文件存放路徑
    :return:
    """
    path = os.path.join(local_data_home, s_table_name)
    if not os.path.exists(path):
        os.makedirs(path, exist_ok=True)
    return path


def copy_to_from_pg(s_table_name):
    """
    從PostgreSQL導出數(shù)據(jù)文件到本地
    :return:
    """
    start = datetime.datetime.now()
    file_prefix = get_file_prefix(s_table_name)
    path = get_local_path(s_table_name)
    full_data_name = os.path.join(path, get_file_name(file_prefix))
    columns = get_column(s_table_name)

    conn = None
    try:
        conn = get_conn('SOURCE')
        if conn is None:
            raise Exception('獲取數(shù)據(jù)庫連接失敗')
        logger.debug(full_data_name)
        with conn.cursor() as cur:
            with open(full_data_name, mode='w', encoding='utf-8') as fileObj:
                cur.copy_to(fileObj, s_table_name, null='NULL', columns=columns)
    finally:
        if conn:
            conn.close()
        end = datetime.datetime.now()
        s = (end - start).total_seconds()
        logger.info('數(shù)據(jù)導出: %s, 耗時: %s 秒' % (s_table_name, s))


def copy_from(s_table_name):
    """
    從本地導入數(shù)據(jù)文件到本地數(shù)據(jù)庫
    :return:
    """
    start = datetime.datetime.now()
    file_prefix = get_file_prefix(s_table_name)
    path = get_local_path(s_table_name)
    full_data_name = os.path.join(path, get_file_name(file_prefix))
    conn = None
    try:
        conn = get_conn('LOCAL')
        with conn:
            # 數(shù)據(jù)文件導入
            sql = "TRUNCATE TABLE %s" % s_table_name
            with conn.cursor() as cur:
                cur.execute(sql)
            with conn.cursor() as cur:
                with open(full_data_name, mode='r', encoding='utf-8') as fileObj:
                    cur.copy_from(fileObj, s_table_name, null='NULL')

    finally:
        if conn:
            conn.close()
        end = datetime.datetime.now()
        s = (end - start).total_seconds()
        logger.info('數(shù)據(jù)導入: %s, 耗時: %s 秒' % (s_table_name, s))


def copy_deal():
    s_table_name = 'public.dim_emp'
    # 從PostgreSQL導出數(shù)據(jù)文件到本地
    copy_to_from_pg(s_table_name)
    # 從本地導入數(shù)據(jù)文件到銀聯(lián)數(shù)據(jù)庫
    copy_from(s_table_name)


if __name__ == '__main__':
    copy_deal()

• settings.py

import os.path
import logging.handlers

BASE_DIR = '/home/xsl/test/'

db_param = {
    "LOCAL": {
        'host': '10.0.0.01',
        'port': 5432,
        'dbname': 'test',    
        'user': 'test01',
        'password': 'Test01',
        'DBType': 'PostgreSQL',
        'remark': '本地數(shù)據(jù)庫',
    },
    "SOURCE": {
         'host': '10.0.0.02',
        'port': 5432,
        'dbname': 'test',    
        'user': 'test02',
        'password': 'Test02',
        'DBType': 'PostgreSQL',
        'remark': '源數(shù)據(jù)庫',
    }
}

log_level = logging.DEBUG
# 日志文件目錄
log_home = os.path.join(BASE_DIR, 'log', 'test')
print(log_home)
if not os.path.exists(log_home):
    os.makedirs(log_home, exist_ok=True)

log_config = {
    'version': 1,
    'formatters': {
        'generic': {
            'format': '%(asctime)s %(levelname)-5.5s [%(name)s:%(lineno)s][%(threadName)s] %(message)s',
        },
        'simple': {
            'format': '%(asctime)s %(levelname)-5.5s %(message)s',
        },
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'formatter': 'generic',
        },
        'file': {
            'class': 'logging.FileHandler',
            'filename': os.path.join(log_home, 'test.log'),
            'encoding': 'utf-8',
            'formatter': 'generic',

        },
    },
    'root': {
        'level': log_level,
        'handlers': ['console', 'file'],
    }
}

# 數(shù)據(jù)文件目錄
local_data_home = os.path.join(BASE_DIR, 'data')
if not os.path.exists(local_data_home):
    os.makedirs(local_data_home, exist_ok=True)