db_sync_tool.database.utility.get_database_tables_like() - Code Metrics - Inspection of "[Release] Version 2.5.5" - jackd248/db-sync-tool - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( f57091...64c755 )

by Konrad

created 2021-10-08 07:12 UTC

get_database_tables_like() A

↳ Parent: db_sync_tool.database.utility

Complexity

Conditions

Size

Total Lines	12
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	6
dl	0
loc	12
rs	10
c	0
b	0
f	0
cc	2
nop	2

#!/usr/bin/env python3
# -*- coding: future_fstrings -*-

"""
Utility script
"""

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

database_dump_file_name = None



class DatabaseSystem:

    MYSQL = 'MySQL'
    MARIADB = 'MariaDB'


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


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


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


def generate_ignore_database_tables():
    """
    Generate the ignore tables options for the mysqldump command by the given table list
    # ToDo: Too much conditional nesting
    :return: String
    """
    _ignore_tables = []
    if 'ignore_table' in system.config:
        for table in system.config['ignore_table']:
            if '*' in table:
                _wildcard_tables = get_database_tables_like(mode.Client.ORIGIN,
                                                            table.replace('*', ''))
                if _wildcard_tables:
                    for wildcard_table in _wildcard_tables:
                        _ignore_tables = generate_ignore_database_table(_ignore_tables,
                                                                        wildcard_table)
            else:
                _ignore_tables = generate_ignore_database_table(_ignore_tables, table)
        return ' '.join(_ignore_tables)
    return ''


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


def get_database_tables_like(client, name):

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


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


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.config['check_dump']:
        _line = mode.run_command(
            helper.get_command(client, 'tail') + ' -n 1 ' + filepath,
            client,
            True,
            skip_dry_run=True
        )

        if not _line:
            return

        if "-- Dump completed on" not in _line:
            sys.exit(
                output.message(
                    output.Subject.ERROR,
                    'Dump file is corrupted',
                    do_print=False
                )
            )
        else:
            output.message(
                output.host_to_subject(client),
                'Dump file is valid',
                verbose_only=True
            )


def count_tables(client, filepath):
    """
    Count the reference string in the database dump file to get the count of all exported tables
    :param client: String
    :param filepath: String
    :return:
    """
    _reference = 'CREATE TABLE'
    _count = mode.run_command(
        f'{helper.get_command(client, "grep")} -ao "{_reference}" {filepath} | wc -l | xargs',
        client,
        True,
        skip_dry_run=True
    )

    if _count:
        output.message(
            output.host_to_subject(client),
            f'{int(_count)} table(s) exported'
        )


def get_database_version(client):
    """
    Check the database version and distinguish between mysql and mariadb
    :param client:
    :return: Tuple<String,String>
    """
    try:
        _database_version = run_database_command(mode.Client.ORIGIN, 'SELECT VERSION();').splitlines()[1]

        _database_system = DatabaseSystem.MYSQL

        _version_number = re.search('(\d+\.)?(\d+\.)?(\*|\d+)', _database_version).group()


        if DatabaseSystem.MARIADB.lower() in _database_version.lower():
            _database_system = DatabaseSystem.MARIADB

        output.message(
            output.host_to_subject(client),
            f'Database version: {_database_system} v{_version_number}',
            True
        )
        return _database_system, _version_number
    finally:
        return None





1			#!/usr/bin/env python3
2			# -- coding: future_fstrings --
3
4			"""
5			Utility script
6			"""
7
8			import sys
9			import datetime
10			import re
11			from db_sync_tool.utility import mode, system, helper, output
12
13			database_dump_file_name = None
			0 ignored issues – show Coding Style Naming introduced 2020-11-27 11:55 UTC by Report Bug Copy Issue Report Constant name "database_dump_file_name" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]\|__.__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
14
15
16			class DatabaseSystem:
			0 ignored issues – show introduced 2021-10-08 07:21 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
17			MYSQL = 'MySQL'
18			MARIADB = 'MariaDB'
19
20
21			def run_database_command(client, command):
22			"""
23			Run a database command using the "mysql -e" command
24			:param client: String
25			:param command: String database command
26			:return:
27			"""
28			return mode.run_command(
29			helper.get_command(client, 'mysql') + ' ' + generate_mysql_credentials(
30			client) + ' -e "' + command + '"',
31			client, True)
32
33
34			def generate_database_dump_filename():
35			"""
36			Generate a database dump filename like "_[name]_[date].sql" or using the give filename
37			:return:
38			"""
39			global database_dump_file_name
			0 ignored issues – show Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Usage of the `global` statement should be avoided. Usage of `global` can make code hard to read and test, its usage is generally not recommended unless you are dealing with legacy code. Loading history... Coding Style Naming introduced 2020-11-27 11:55 UTC by Report Bug Copy Issue Report Constant name "database_dump_file_name" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\Wa-z]\|__.__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
40
41			if system.config['dump_name'] == '':
42			# _project-db_20-08-2020_12-37.sql
43			_now = datetime.datetime.now()
44			database_dump_file_name = '_' + system.config['origin']['db']['name'] + '_' + _now.strftime(
45			"%d-%m-%Y_%H-%M") + '.sql'
46			else:
47			database_dump_file_name = system.config['dump_name'] + '.sql'
48
49
50			def generate_ignore_database_tables():
51			"""
52			Generate the ignore tables options for the mysqldump command by the given table list
53			# ToDo: Too much conditional nesting
54			:return: String
55			"""
56			_ignore_tables = []
57			if 'ignore_table' in system.config:
58			for table in system.config['ignore_table']:
59			if '*' in table:
60			_wildcard_tables = get_database_tables_like(mode.Client.ORIGIN,
61			table.replace('*', ''))
62			if _wildcard_tables:
63			for wildcard_table in _wildcard_tables:
64			_ignore_tables = generate_ignore_database_table(_ignore_tables,
65			wildcard_table)
66			else:
67			_ignore_tables = generate_ignore_database_table(_ignore_tables, table)
68			return ' '.join(_ignore_tables)
69			return ''
70
71
72			def generate_ignore_database_table(ignore_tables, table):
73			"""
74			:param ignore_tables: Dictionary
75			:param table: String
76			:return: Dictionary
77			"""
78			ignore_tables.append('--ignore-table=' + system.config['origin']['db']['name'] + '.' + table)
79			return ignore_tables
80
81
82			def get_database_tables_like(client, name):
			0 ignored issues – show Unused Code introduced 2020-11-26 10:20 UTC by Report Bug Copy Issue Report Either all return statements in a function should return an expression, or none of them should. Loading history...
83			"""
84			Get database table names like the given name
85			:param client: String
86			:param name: String
87			:return: Dictionary
88			"""
89			_dbname = system.config[client]['db']['name']
90			_tables = run_database_command(client, f'SHOW TABLES FROM {_dbname} LIKE \'%{name}%\';').strip()
91			if _tables != '':
92			return _tables.split('\n')[1:]
93			return
94
95
96			def generate_mysql_credentials(client):
97			"""
98			Generate the needed database credential information for the mysql command
99			:param client: String
100			:return:
101			"""
102			_credentials = '-u\'' + system.config[client]['db']['user'] + '\' -p\'' + \
103			system.config[client]['db'][
104			'password'] + '\''
105			if 'host' in system.config[client]['db']:
106			_credentials += ' -h\'' + system.config[client]['db']['host'] + '\''
107			if 'port' in system.config[client]['db']:
108			_credentials += ' -P\'' + str(system.config[client]['db']['port']) + '\''
109			return _credentials
110
111
112			def check_database_dump(client, filepath):
113			"""
114			Checking the last line of the dump file if it contains "-- Dump completed on"
115			:param client: String
116			:param filepath: String
117			:return:
118			"""
119			if system.config['check_dump']:
120			_line = mode.run_command(
121			helper.get_command(client, 'tail') + ' -n 1 ' + filepath,
122			client,
123			True,
124			skip_dry_run=True
125			)
126
127			if not _line:
128			return
129
130			if "-- Dump completed on" not in _line:
131			sys.exit(
132			output.message(
133			output.Subject.ERROR,
134			'Dump file is corrupted',
135			do_print=False
136			)
137			)
138			else:
139			output.message(
140			output.host_to_subject(client),
141			'Dump file is valid',
142			verbose_only=True
143			)
144
145
146			def count_tables(client, filepath):
147			"""
148			Count the reference string in the database dump file to get the count of all exported tables
149			:param client: String
150			:param filepath: String
151			:return:
152			"""
153			_reference = 'CREATE TABLE'
154			_count = mode.run_command(
155			f'{helper.get_command(client, "grep")} -ao "{_reference}" {filepath} \| wc -l \| xargs',
156			client,
157			True,
158			skip_dry_run=True
159			)
160
161			if _count:
162			output.message(
163			output.host_to_subject(client),
164			f'{int(_count)} table(s) exported'
165			)
166
167
168			def get_database_version(client):
169			"""
170			Check the database version and distinguish between mysql and mariadb
171			:param client:
172			:return: Tuple<String,String>
173			"""
174			try:
175			_database_version = run_database_command(mode.Client.ORIGIN, 'SELECT VERSION();').splitlines()[1]
			0 ignored issues – show Coding Style introduced 2021-10-08 07:21 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (105/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
176			_database_system = DatabaseSystem.MYSQL
177
178			_version_number = re.search('(\d+\.)?(\d+\.)?(\*\|\d+)', _database_version).group()
			0 ignored issues – show Bug introduced 2021-10-08 07:21 UTC by Report Bug Copy Issue Report A suspicious escape sequence `\.` was found. Did you maybe forget to add an `r` prefix? Escape sequences in Python are generally interpreted according to rules similar to standard C. Only if strings are prefixed with `r` or `R` are they interpreted as regular expressions. The escape sequence that was used indicates that you might have intended to write a regular expression. Learn more about the available escape sequences. in the Python documentation. Loading history... Bug introduced 2021-10-08 07:21 UTC by Report Bug Copy Issue Report A suspicious escape sequence `\d` was found. Did you maybe forget to add an `r` prefix? Escape sequences in Python are generally interpreted according to rules similar to standard C. Only if strings are prefixed with `r` or `R` are they interpreted as regular expressions. The escape sequence that was used indicates that you might have intended to write a regular expression. Learn more about the available escape sequences. in the Python documentation. Loading history... Bug introduced 2021-10-08 07:21 UTC by Report Bug Copy Issue Report A suspicious escape sequence `\*` was found. Did you maybe forget to add an `r` prefix? Escape sequences in Python are generally interpreted according to rules similar to standard C. Only if strings are prefixed with `r` or `R` are they interpreted as regular expressions. The escape sequence that was used indicates that you might have intended to write a regular expression. Learn more about the available escape sequences. in the Python documentation. Loading history...
179
180			if DatabaseSystem.MARIADB.lower() in _database_version.lower():
181			_database_system = DatabaseSystem.MARIADB
182
183			output.message(
184			output.host_to_subject(client),
185			f'Database version: {_database_system} v{_version_number}',
186			True
187			)
188			return _database_system, _version_number
189			finally:
190			return None
			0 ignored issues – show Bug Best Practice introduced 2021-10-08 07:21 UTC by Report Bug Copy Issue Report `return` statements in `finally` blocks should be avoided. Placing a return statement inside finally will swallow all exceptions that may have been thrown in the try block. Loading history...
191
			0 ignored issues – show coding-style introduced 2021-10-08 07:21 UTC by Report Bug Copy Issue Report Trailing newlines Loading history...
192

jackd248 / db-sync-tool

Push — master ( f57091...64c755 )

get_database_tables_like() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like