ICode9

精准搜索请尝试: 精确搜索
首页 > 其他分享> 文章详细

跨库数据备份还原、迁移工具

2022-03-06 11:34:46  阅读:190  来源: 互联网

标签:name 跨库 数据库 数据备份 table path 迁移 os conn


前言

数据开发过程中,为了确保生产数据库安全,一般将实时数据同步、备份到本地测试数据库完成开发工作,最后部署应用。

本文实际业务场景:需要从客户处的生成数据库(如mysql数据库)备份数据表到本地的一个sqlserver库中,将生产数据库表复制还原过来。为了快速完成数据备份操作,于是我写了个同步迁移的代码。同样的使用需求下,只需修改配置settings就可以直接用。

当然也可以用数据库客户端界面可视化完成数据库的备份还原。

测试数据准备

本文还是利用上一篇文章的数据表,该数据存放在mysql数据库中。

Python搭建一个系统信息实时监控数据可视化大屏

先查看system_info表结构。

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从文件复制数据到插入到目标数据库表中。

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:\myProjectfile\database_backup\logs'

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)
同步完成

 

 

 

 

转自:微信公众号: Python数据分析实例

原文:https://mp.weixin.qq.com/s?__biz=MzI0NzY2MDA4MA==&mid=2247498197&idx=1&sn=4d0ef699f48482e48c1ed2c111efad3c&scene=21#wechat_redirect

标签:name,跨库,数据库,数据备份,table,path,迁移,os,conn
来源: https://www.cnblogs.com/tjp40922/p/15971137.html

本站声明: 1. iCode9 技术分享网(下文简称本站)提供的所有内容,仅供技术学习、探讨和分享;
2. 关于本站的所有留言、评论、转载及引用,纯属内容发起人的个人观点,与本站观点和立场无关;
3. 关于本站的所有言论和文字,纯属内容发起人的个人观点,与本站观点和立场无关;
4. 本站文章均是网友提供,不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属;如您发现该文章侵犯了您的权益,可联系我们第一时间进行删除;
5. 本站为非盈利性的个人网站,所有内容不会用来进行牟利,也不会利用任何形式的广告来间接获益,纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

专注分享技术,共同学习,共同进步。侵权联系[81616952@qq.com]

Copyright (C)ICode9.com, All Rights Reserved.

ICode9版权所有