前言
数据开发过程中，为了确保生产数据库安全，一般将实时数据同步、备份到本地测试数据库完成开发工作，最后部署应用。
本文实际业务场景：需要从客户处的生成数据库（如mysql数据库）备份数据表到本地的一个sqlserver库中，将生产数据库表复制还原过来。为了快速完成数据备份操作，于是我写了个同步迁移的代码。同样的使用需求下，只需修改配置settings就可以直接用。
当然也可以用数据库客户端界面可视化完成数据库的备份还原。
测试数据准备
本文还是利用上一篇文章的数据表，该数据存放在mysql数据库中。
Python搭建一个系统信息实时监控数据可视化大屏
先查看system_info表结构。
代码语言：javascript复制SHOW FULL COLUMNS FROM system_info 
代码结构
主函数db_backup.py完成数据库表同库、垮库的备份、同步、迁移功能，logging_conf.py日志记录模块；settings.py设置两个数据库配置参数；tools.py完成路径获取和数据库连接功能。
代码部分
- 数据库备份、迁移-db_backup.py
该模块主要方法：copy_to_from_mysql和mysql_sync_to_sqlserver。
1、copy_to_from_mysql用于把一个表的内容复制到一个文件。
2、mysql_sync_to_sqlserver从文件复制数据到插入到目标数据库表中。
代码语言：javascript复制
import datetime
import logging.config
from tools import get_local_path,get_conn
from logging_conf import log_config,local_data
logging.config.dictConfig(log_config)
logger = logging.getLogger(__name__)
def copy_to_from_mysql(table_name):
"""
从Mysql导出数据文件到本地
"""
start = datetime.datetime.now()
full_data_name = get_local_path(table_name)
conn = None
try:
conn = get_conn('SOURCE')
if conn is None:
raise Exception('获取数据库连接失败')
logger.debug(full_data_name)
sql = 'Select * from {0}'.format(table_name)
with conn.cursor() as cur:
cur.execute(sql)
number = cur.fetchall()
loan_count = 0
for loanNumber in number:
tuple_list = tuple([str(i) for i in list(loanNumber)][1:])
loan_count = 1
with open(full_data_name, mode='a', encoding='utf-8') as f:
f.write(str(tuple_list) "n")
f.close()
cur.close()
print("写入完成,共写入%d条数据！" % loan_count)
finally:
if conn:
conn.close()
end = datetime.datetime.now()
s = (end - start).total_seconds()
logger.info('数据导出: %s, 耗时: %s 秒' % (table_name, s))
return number
def mysql_sync_to_sqlserver(table_name):
"""
把mysql数据同步到sqlserver数据库里面
从本地导入数据文件到本地数据库
:return:
"""
start = datetime.datetime.now()
full_data_name = get_local_path(table_name)
conn = None
try:
conn = get_conn('LOCAL')
with conn:
# 数据文件导入
with conn.cursor() as cur:
with open(full_data_name, mode='r', encoding='utf-8') as lst:
for line in lst:
sql = "insert into system_info values {0}".format(line)
print("插入成功",line)
cur.execute(sql)
conn.commit()
finally:
if conn:
conn.close()
end = datetime.datetime.now()
s = (end - start).total_seconds()
logger.info('数据导入: %s, 耗时: %s 秒' % (table_name, s))
if __name__ == '__main__':
table_name = 'system_info'
# 从mysql导出数据文件到本地
copy_to_from_mysql(table_name)
# 从本地导入数据文件到sqlserver数据库
mysql_sync_to_sqlserver(table_name)
- 数据库参数配置-settings.py 填写数据库备份、还原数据库相关配置参数
db_param = {
"LOCAL": {
'host': 'localhost',
'port': 1433,
'dbname': 'test',
'user': 'xxxx',
'password': 'xxxx',
'DBType': 'SQLServer',
'remark': '本地数据库',
},
"SOURCE": {
'host': 'localhost',
'port':3306,
'dbname': 'mydb',
'user': 'xxxx',
'password': 'xxxx',
'DBType': 'Mysql',
'remark': '目标数据库',
}
}
- 日志记录模块-logging_conf.py
import os.path
import logging.handlers
BASE_DIR = r'D:myProjectfiledatabase_backuplogs'
log_level = logging.DEBUG
# 日志文件位置
log_home = os.path.join(BASE_DIR, 'log', 'test')
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 = os.path.join(BASE_DIR, 'data')
if not os.path.exists(local_data):
os.makedirs(local_data, exist_ok=True)
- 数据库连接和路径获取工具-tools.py
import os
import pymysql
import pymssql
from settings import db_param
from logging_conf import local_data
def get_file_name(table_name, suffix='txt'):
"""
返回文件名
"""
return table_name.lower() '.' suffix
def get_local_path(table_name):
"""
本地文件存放路径
"""
path = os.path.join(local_data, table_name)
if not os.path.exists(path):
os.makedirs(path, exist_ok=True)
full_data_name = os.path.join(path, get_file_name(table_name))
return full_data_name
def get_conn(sys_code='SOURCE'):
"""
数据库连接获取，此处给出我常用的三种数据库连接
"""
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 == "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")
else:
raise Exception("%s数据库连接失败. " % sys_code)
同步完成
