最近遇到一個場景需要從一個postgresql庫同步一張表到另一個postgresql庫中，但又不需要實時同步到踏，就寫了個同步的代碼杠袱，本來網(wǎng)上同步的方法早都有了，之所以自己寫一套窝稿，是因為postgresql數(shù)據(jù)庫可用的太少了楣富，于是我決定擼起袖子再寫一套。
整個代碼部分就不再過多啰嗦了伴榔，因為都是一些基礎(chǔ)纹蝴，目的只有一個：讓你快速可以使用。如果有同樣的需求踪少，改下配置settings就可以直接用塘安。如果的確有看不懂的地方，請把你的疑惑留在評論區(qū)援奢，如果沒有兼犯，那我的目的就達到了。整塊代碼主要用到兩個方法copy_to集漾、copy_from
-
copy_to用于把一個表的內(nèi)容復(fù)制到一個文件切黔；copy_to中也可以指定查詢，將查詢結(jié)果寫入文件 -
copy_from從文件復(fù)制數(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：要被復(fù)制的列列表
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導(dǎo)出數(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ù)導(dǎo)出: %s, 耗時: %s 秒' % (s_table_name, s))
def copy_from(s_table_name):
"""
從本地導(dǎo)入數(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ù)文件導(dǎo)入
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ù)導(dǎo)入: %s, 耗時: %s 秒' % (s_table_name, s))
def copy_deal():
s_table_name = 'public.dim_emp'
# 從PostgreSQL導(dǎo)出數(shù)據(jù)文件到本地
copy_to_from_pg(s_table_name)
# 從本地導(dǎo)入數(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)
