你有没有遇到需要跨库同步数据的？

最近遇到一个场景需要从一个postgresql库同步一张表到另一个postgresql库中，但又不需要实时同步，就写了个同步的代码，本来网上同步的方法早都有了，之所以自己写一套，是因为postgresql数据库可用的太少了，于是我决定撸起袖子再写一套。

整个代码部分就不再过多啰嗦了，因为都是一些基础，目的只有一个：让你快速可以使用。如果有同样的需求，改下配置settings就可以直接用。如果的确有看不懂的地方，请把你的疑惑留在评论区，如果没有，那我的目的就达到了。整块代码主要用到两个方法

copy_to

、

copy_from

• copy_to

  用于把一个表的内容复制到一个文件；

  copy_to

  中也可以指定查询，将查询结果写入文件

• copy_from

  从文件复制数据到表中。

  copy_from

  中，文件的字段按照顺序写入到指定列中。

需要注意的是：

1.数据库用户必须有文件所在的路径的写权限。

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'):
    """
    数据库连接获取
    :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("源系统%s数据库连接失败. " % 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导出数据文件到本地
    :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('获取数据库连接失败')
        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('数据导出: %s, 耗时: %s 秒' % (s_table_name, s))


def copy_from(s_table_name):
    """
    从本地导入数据文件到本地数据库
    :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:
            # 数据文件导入
            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('数据导入: %s, 耗时: %s 秒' % (s_table_name, s))


def copy_deal():
    s_table_name = 'public.dim_emp'
    # 从PostgreSQL导出数据文件到本地
    copy_to_from_pg(s_table_name)
    # 从本地导入数据文件到银联数据库
    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': '本地数据库',
    },
    "SOURCE": {
         'host': '10.0.0.02',
        'port': 5432,
        'dbname': 'test',    
        'user': 'test02',
        'password': 'Test02',
        'DBType': 'PostgreSQL',
        'remark': '源数据库',
    }
}

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'],
    }
}

# 数据文件目录
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)           

复制