preface

In the process of data development, in order to ensure the safety of the production database, the real-time data is generally synchronized and backed up to the local test database to complete the development work, and finally deploy the application.

The actual business scenario of this article: you need to copy and restore the production database table from the backup data table of the customer's generated database (such as mysql database) to a local SQL Server database. In order to quickly complete the data backup operation, I wrote a code for synchronous migration. Under the same use requirements, you can use it directly by modifying the configuration settings.

Of course, you can also use the database client interface to visually complete the backup and restore of the database.

Test data preparation

This paper also uses the data table of the previous article, which is stored in mysql database.

Python builds a large visualization screen for real-time monitoring data of system information

Check the system first_ Info table structure.

SHOW FULL COLUMNS FROM system_info 

Code structure

Main function db_backup.py completes the backup, synchronization and migration functions of database tables, the same database and broken database, and logging_conf.py logging module; settings.py set two database configuration parameters; tools.py completes the functions of path acquisition and database connection.

Code part

• Database backup and migration - db_backup.py

The main method of this module: copy_to_from_mysql and mysql_sync_to_sqlserver.  

1,copy_to_from_mysql is used to copy the contents of a table to a file.

2,mysql_ sync_ to_ SQL Server copies data from the file and inserts it into the target database table.

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):
    """
    from Mysql Export data file to local
    """
    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('Failed to get database connection')
        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("Write complete,Co write%d Data!" % loan_count)

    finally:
        if conn:
            conn.close()
        end = datetime.datetime.now()
        s = (end - start).total_seconds()
        logger.info('Data export: %s, time consuming: %s second' % (table_name, s))

    return number

def mysql_sync_to_sqlserver(table_name):
    """
    hold mysql Data synchronization to sqlserver In the database
    Import data files from local to local database
    :return:
    """
    start = datetime.datetime.now()
    full_data_name = get_local_path(table_name)
    conn = None
    try:
        conn = get_conn('LOCAL')
        with conn:
            # Data file import
            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("Insert successful",line)
                        cur.execute(sql)
            conn.commit()

    finally:
        if conn:
            conn.close()
        end = datetime.datetime.now()
        s = (end - start).total_seconds()
        logger.info('Data import: %s, time consuming: %s second' % (table_name, s))


if __name__ == '__main__':
    table_name = 'system_info'
    # from mysql Export data file to local
    copy_to_from_mysql(table_name)
    # Import data files from local to sqlserver database
    mysql_sync_to_sqlserver(table_name) 

• Database parameter configuration - settings py

  Fill in the configuration parameters related to database backup and restore

db_param = {
    "LOCAL": {
        'host': 'localhost',
        'port': 1433,
        'dbname': 'test',
        'user': 'xxxx',
        'password': 'xxxx',
        'DBType': 'SQLServer',
        'remark': 'Local database',
    },
    "SOURCE": {
         'host': 'localhost',
        'port':3306,
        'dbname': 'mydb',
        'user': 'xxxx',
        'password': 'xxxx',
        'DBType': 'Mysql',
        'remark': 'Target database',
    }
}

• Logging module - logging_conf.py

import os.path
import logging.handlers

BASE_DIR = r'D:\myProjectfile\database_backup\logs'

log_level = logging.DEBUG

# Log file location
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',   #Output to terminal
            'formatter': 'generic',
        },
        'file': {
            'class': 'logging.FileHandler',  #output to a file
            'filename': os.path.join(log_home, 'test.log'),
            'encoding': 'utf-8',
            'formatter': 'generic',

        },
    },
    'root': {
        'level': log_level,
        'handlers': ['console', 'file'],
    }
}

# Data file location (temporary data storage location)
local_data = os.path.join(BASE_DIR, 'data')
if not os.path.exists(local_data):
    os.makedirs(local_data, exist_ok=True)

• Database connection and path acquisition tool - 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 file name
    """
    return table_name.lower() + '.' + suffix

def get_local_path(table_name):
    """
    Local file storage path
    """
    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'):
    """
    Database connection acquisition. Here are three kinds of database connections I often use
    """
    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 Database connection failed. " % sys_code)

Synchronization complete

 

 

 

 

Transfer from: WeChat official account: Python data analysis example

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