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

Passed

Push — master ( 7fdd2b...425778 )

by Konrad

created 2022-03-06 12:53 UTC

get_database_tables() A

↳ Parent: db_sync_tool.database.utility

Complexity

Conditions

Size

Total Lines	13
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	8
dl	0
loc	13
rs	10
c	0
b	0
f	0
cc	3
nop	0

#!/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 get_database_tables():
    """
    Generate specific tables for export
    :return: String
    """
    if system.config['tables'] == '':
        return ''

    _result = ''
    _tables = system.config['tables'].split(',')
    for _table in _tables:
        _result += '\'' + _table + '\' '
    return _result


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>
    """
    _database_system = None
    _version_number = None
    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
        )
    finally:
        return _database_system, _version_number





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 get_database_tables():
97			"""
98			Generate specific tables for export
99			:return: String
100			"""
101			if system.config['tables'] == '':
102			return ''
103
104			_result = ''
105			_tables = system.config['tables'].split(',')
106			for _table in _tables:
107			_result += '\'' + _table + '\' '
108			return _result
109
110
111			def generate_mysql_credentials(client):
112			"""
113			Generate the needed database credential information for the mysql command
114			:param client: String
115			:return:
116			"""
117			_credentials = '-u\'' + system.config[client]['db']['user'] + '\' -p\'' + \
118			system.config[client]['db'][
119			'password'] + '\''
120			if 'host' in system.config[client]['db']:
121			_credentials += ' -h\'' + system.config[client]['db']['host'] + '\''
122			if 'port' in system.config[client]['db']:
123			_credentials += ' -P\'' + str(system.config[client]['db']['port']) + '\''
124			return _credentials
125
126
127			def check_database_dump(client, filepath):
128			"""
129			Checking the last line of the dump file if it contains "-- Dump completed on"
130			:param client: String
131			:param filepath: String
132			:return:
133			"""
134			if system.config['check_dump']:
135			_line = mode.run_command(
136			helper.get_command(client, 'tail') + ' -n 1 ' + filepath,
137			client,
138			True,
139			skip_dry_run=True
140			)
141
142			if not _line:
143			return
144
145			if "-- Dump completed on" not in _line:
146			sys.exit(
147			output.message(
148			output.Subject.ERROR,
149			'Dump file is corrupted',
150			do_print=False
151			)
152			)
153			else:
154			output.message(
155			output.host_to_subject(client),
156			'Dump file is valid',
157			verbose_only=True
158			)
159
160
161			def count_tables(client, filepath):
162			"""
163			Count the reference string in the database dump file to get the count of all exported tables
164			:param client: String
165			:param filepath: String
166			:return:
167			"""
168			_reference = 'CREATE TABLE'
169			_count = mode.run_command(
170			f'{helper.get_command(client, "grep")} -ao "{_reference}" {filepath} \| wc -l \| xargs',
171			client,
172			True,
173			skip_dry_run=True
174			)
175
176			if _count:
177			output.message(
178			output.host_to_subject(client),
179			f'{int(_count)} table(s) exported'
180			)
181
182
183			def get_database_version(client):
184			"""
185			Check the database version and distinguish between mysql and mariadb
186			:param client:
187			:return: Tuple<String,String>
188			"""
189			_database_system = None
190			_version_number = None
191			try:
192			_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...
193			_database_system = DatabaseSystem.MYSQL
194
195			_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 `\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... 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...
196
197			if DatabaseSystem.MARIADB.lower() in _database_version.lower():
198			_database_system = DatabaseSystem.MARIADB
199
200			output.message(
201			output.host_to_subject(client),
202			f'Database version: {_database_system} v{_version_number}',
203			True
204			)
205			finally:
206			return _database_system, _version_number
			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...
207
			0 ignored issues – show coding-style introduced 2021-10-08 07:21 UTC by Report Bug Copy Issue Report Trailing newlines Loading history...
208

jackd248 / db-sync-tool

Push — master ( 7fdd2b...425778 )

get_database_tables() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like