db_sync_tool.utility.database

Complexity

Total Complexity

28

Size/Duplication

Total Lines	208
Duplicated Lines	0 %

Importance

Changes

0

11 Functions

Rating	Name	Duplication	Size	Complexity
A	create_origin_database_dump()	0	27	2
A	run_database_command()	0	7	1
A	get_database_tables_like()	0	11	2
B	generate_ignore_database_tables()	0	16	6
A	generate_mysql_credentials()	0	13	3
A	generate_ignore_database_table()	0	8	1
A	prepare_target_database_dump()	0	11	1
A	check_database_dump()	0	17	3
A	prepare_origin_database_dump()	0	15	1
B	import_database_dump()	0	26	6
A	generate_database_dump_filename()	0	13	2

#!/usr/bin/env python3

Missing module docstring

# -*- coding: utf-8 -*-

import sys
import datetime
from db_sync_tool.utility import parser, mode, system, helper, output

#
# GLOBALS
#

origin_database_dump_file_name = None

Constant name "origin_database_dump_file_name" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]*|__.*__)$' pattern)

#
# FUNCTIONS
#

def create_origin_database_dump():
    """
    Creating the origin database dump file
    :return:
    """

    if not mode.is_import():
        parser.get_database_configuration(mode.Client.ORIGIN)
        generate_database_dump_filename()
        helper.check_and_create_dump_dir(mode.Client.ORIGIN, helper.get_dump_dir(mode.Client.ORIGIN))

This line is too long as per the coding-style (101/100).

        _dump_file_path = helper.get_dump_dir(mode.Client.ORIGIN) + origin_database_dump_file_name

        output.message(
            output.Subject.ORIGIN,
            'Creating database dump',
            True
        )
        mode.run_command(
            helper.get_command('origin', 'mysqldump') + ' --no-tablespaces ' + generate_mysql_credentials('origin') + ' ' +

This line is too long as per the coding-style (123/100).

            system.config['db']['origin'][
                'dbname'] + ' ' + generate_ignore_database_tables() + ' > ' + _dump_file_path,
            mode.Client.ORIGIN
        )

        check_database_dump(mode.Client.ORIGIN, _dump_file_path)
        prepare_origin_database_dump()


def run_database_command(client, command):
    """
    Run a database command using the "mysql -e" command
    :param command: String database command
    :return:
    """
    return mode.run_command(helper.get_command(client, 'mysql') + ' ' + generate_mysql_credentials(client) + ' -e "' + command + '"', client, True)

This line is too long as per the coding-style (147/100).

def generate_database_dump_filename():
    """
    Generate a database dump filename like "_[dbname]_[date].sql" or using the give filename
    :return:
    """
    global origin_database_dump_file_name

Usage of the global statement should be avoided.

Constant name "origin_database_dump_file_name" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]*|__.*__)$' pattern)

    if system.option['dump_name'] == '':
        # _project-db_20-08-2020_12-37.sql
        _now = datetime.datetime.now()
        origin_database_dump_file_name = '_' + system.config['db']['origin']['dbname'] + '_' + _now.strftime("%d-%m-%Y_%H-%M") + '.sql'

This line is too long as per the coding-style (135/100).

    else:
        origin_database_dump_file_name = system.option['dump_name'] + '.sql'


def generate_ignore_database_tables():

Either all return statements in a function should return an expression, or none of them should.

    """
    Generate the ignore tables options for the mysqldump command by the given table list
    :return: String
    """
    _ignore_tables = []
    if 'ignore_table' in system.config['host']:
        for table in system.config['host']['ignore_table']:
            if '*' in table:
                _wildcard_tables = get_database_tables_like('origin', table.replace('*', ''))
                if _wildcard_tables:
                    for wildcard_table in _wildcard_tables:
                        _ignore_tables = generate_ignore_database_table(_ignore_tables, wildcard_table)

This line is too long as per the coding-style (103/100).

            else:
                _ignore_tables = generate_ignore_database_table(_ignore_tables, table)
        return ' '.join(_ignore_tables)


def generate_ignore_database_table(ignore_tables, table):
    """
    :param ignore_tables: Dictionary
    :param table: String
    :return:
    """
    ignore_tables.append('--ignore-table=' + system.config['db']['origin']['dbname'] + '.' + table)
    return ignore_tables


def get_database_tables_like(client, name):

Either all return statements in a function should return an expression, or none of them should.

    """
    Get database table names like the given name
    :param client: String
    :param name: String
    :return: Dictionary
    """
    _dbname = system.config['db'][client]['dbname']
    _tables = run_database_command(client, f'SHOW TABLES FROM {_dbname} LIKE \'%{name}%\';').strip()
    if _tables != '':
        return _tables.split('\n', 1)[1:]


def generate_mysql_credentials(client):
    """
    Generate the needed database credential information for the mysql command
    :param client: String
    :return:
    """
    _credentials = '-u\'' + system.config['db'][client]['user'] + '\' -p\'' + system.config['db'][client][

This line is too long as per the coding-style (106/100).

        'password'] + '\''
    if 'host' in system.config['db'][client]:
        _credentials += ' -h\'' + system.config['db'][client]['host'] + '\''
    if 'port' in system.config['db'][client]:
        _credentials += ' -P\'' + str(system.config['db'][client]['port']) + '\''
    return _credentials


#
# IMPORT DATABASE DUMP
#
def import_database_dump():
    """
    Importing the selected database dump file
    :return:
    """
    if (not system.option['is_same_client'] and not mode.is_import()):
        prepare_target_database_dump()

    if not system.option['keep_dump'] and not system.option['is_same_client']:
        output.message(
            output.Subject.TARGET,
            'Importing database dump',
            True
        )

        if not mode.is_import():
           _dump_path = helper.get_dump_dir(mode.Client.TARGET) + origin_database_dump_file_name

The indentation here looks off. 12 spaces were expected, but 11 were found.

        else:
           _dump_path = system.option['import']

The indentation here looks off. 12 spaces were expected, but 11 were found.

        check_database_dump(mode.Client.TARGET, _dump_path)
        mode.run_command(
            helper.get_command('target', 'mysql') + ' ' + generate_mysql_credentials('target') + ' ' +

This line is too long as per the coding-style (102/100).

            system.config['db']['target'][
                'dbname'] + ' < ' + _dump_path,
            mode.Client.TARGET
        )


def prepare_origin_database_dump():
    """
    Preparing the origin database dump file by compressing them as .tar.gz
    :return:
    """
    output.message(
        output.Subject.ORIGIN,
        'Compressing database dump',
        True
    )
    mode.run_command(
        helper.get_command(mode.Client.ORIGIN, 'tar') + ' cfvz ' + helper.get_dump_dir(
            mode.Client.ORIGIN) + origin_database_dump_file_name + '.tar.gz -C ' + helper.get_dump_dir(

This line is too long as per the coding-style (103/100).

            mode.Client.ORIGIN) + ' ' + origin_database_dump_file_name + ' > /dev/null',

Wrong hanging indentation (add 4 spaces).

        mode.Client.ORIGIN
    )


def prepare_target_database_dump():
    """
    Preparing the target database dump by the unpacked .tar.gz file
    :return:
    """
    output.message(output.Subject.TARGET, 'Extracting database dump', True)
    mode.run_command(
        helper.get_command('target', 'tar') + ' xzf ' + helper.get_dump_dir(
            mode.Client.TARGET) + origin_database_dump_file_name + '.tar.gz -C ' + helper.get_dump_dir(

This line is too long as per the coding-style (103/100).

            mode.Client.TARGET) + ' > /dev/null',

Wrong hanging indentation (add 4 spaces).

        mode.Client.TARGET
    )


def check_database_dump(client, filepath):
    """
    Checking the last line of the dump file if it contains "-- Dump completed on"
    :param client: String
    :param filepath: String
    :return:
    """

    if system.option['check_dump']:
        _line = mode.run_command(
            helper.get_command(client, 'tail') + ' -n 1 ' + filepath,
                client,

Wrong hanging indentation (remove 4 spaces).

                True

Wrong hanging indentation (remove 4 spaces).

            )

        if "-- Dump completed on" not in _line:
            sys.exit(output.message(output.Subject.ERROR, 'Dump file is corrupted', False))