最近遇到一个场景需要从一个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)
