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

Passed

Push — master ( 190e21...7fdd2b )

by Konrad

created 2022-02-13 19:28 UTC

db_sync_tool.utility.helper.confirm() C

↳ Parent: db_sync_tool.utility.helper

Complexity

Conditions

Size

Total Lines	38
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	17
dl	0
loc	38
rs	5.9999
c	0
b	0
f	0
cc	10
nop	2

How to fix Complexity

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

"""
Helper script
"""

import shutil
import os
import re
from db_sync_tool.utility import mode, system, output
from db_sync_tool.remote import utility as remote_utility


def clean_up():
    """
    Clean up
    :return:
    """
    if not mode.is_import():
        remote_utility.remove_target_database_dump()
        if mode.get_sync_mode() == mode.SyncMode.PROXY:
            remove_temporary_data_dir()


def remove_temporary_data_dir():
    """
    Remove temporary data directory for storing database dump files
    :return:
    """
    if os.path.exists(system.default_local_sync_path):
        output.message(
            output.Subject.LOCAL,
            'Cleaning up',
            True
        )
        shutil.rmtree(system.default_local_sync_path)


def clean_up_dump_dir(client, path, num=5):
    """
    Clean up the dump directory from old dump files (only affect .sql and .tar.gz files)
    :param client:
    :param path:
    :param num:
    :return:
    """
    # Distinguish stat command on os system (Darwin|Linux)
    if check_os(client).strip() == 'Darwin':
        _command = get_command(client, 'stat') + ' -f "%Sm %N" ' + path + ' | ' + get_command(
            client,
            'sort') + ' -rn | ' + get_command(
            client, 'grep') + ' -E ".tar.gz|.sql"'

    else:
        _command = get_command(client, 'stat') + ' -c "%y %n" ' + path + ' | ' + \
                   get_command(client,'sort') + ' -rn | ' + get_command(client, 'grep') + \

                   ' -E ".tar.gz|.sql"'

    # List files in directory sorted by change date
    _files = mode.run_command(
        _command,
        client,
        True
    ).splitlines()

    for i in range(len(_files)):

        _filename = _files[i].rsplit(' ', 1)[-1]

        # Remove oldest files chosen by keep_dumps count
        if not i < num:
            mode.run_command(
                'rm ' + _filename,
                client
            )


def check_os(client):
    """
    Check which system is running (Linux|Darwin)
    :param client:
    :return:
    """
    return mode.run_command(
        get_command(client, 'uname') + ' -s',
        client,
        True
    )


def get_command(client, command):
    """
    Get command helper for overriding default commands on the given client
    :param client:
    :param command:
    :return: String command
    """
    if 'console' in system.config[client]:
        if command in system.config[client]['console']:
            return system.config[client]['console'][command]
    return command


def get_dump_dir(client):
    """
    Get database dump directory by client
    :param client:
    :return: String path
    """
    if system.config[f'default_{client}_dump_dir']:

        return '/tmp/'
    else:
        return system.config[client]['dump_dir']


def check_and_create_dump_dir(client, path):
    """
    Check if a path exists on the client system and creates the given path if necessary
    :param client:
    :param path:
    :return:
    """
    mode.run_command(
        '[ ! -d "' + path + '" ] && mkdir -p "' + path + '"',
        client
    )


def get_ssh_host_name(client, with_user=False):
    """
    Format ssh host name depending on existing client name
    :param client:
    :param with_user:
    :return:
    """
    if not 'user' in system.config[client] and not 'host' in system.config[client]:
        return ''

    if with_user:
        _host = system.config[client]['user'] + '@' + system.config[client]['host']
    else:
        _host = system.config[client]['host']

    if 'name' in system.config[client]:

        return output.CliFormat.BOLD + system.config[client][
            'name'] + output.CliFormat.ENDC + output.CliFormat.BLACK + ' (' + _host + ')' + \
               output.CliFormat.ENDC
    else:
        return _host


def create_local_temporary_data_dir():
    """
    Create local temporary data dir
    :return:
    """
    # @ToDo: Combine with check_and_create_dump_dir()
    if not os.path.exists(system.default_local_sync_path):
        os.makedirs(system.default_local_sync_path)


def dict_to_args(dict):

    """
    Convert an dictionary to a args list
    :param dict: Dictionary
    :return: List
    """
    _args = []
    for key, val in dict.items():
        if isinstance(val, bool):
            if val:
                _args.append(f'--{key}')
        else:
            _args.append(f'--{key}')
            _args.append(str(val))
    if len(_args) == 0:
        return None
    return _args


def check_file_exists(client, path):
    """
    Check if a file exists
    :param client: String
    :param path: String file path
    :return: Boolean
    """
    return mode.run_command(f'[ -f {path} ] && echo "1"', client, True) == '1'


def run_script(client=None, script='before'):
    """
    Executing script command
    :param client: String
    :param script: String
    :return:
    """
    if client is None:
        _config = system.config
        _subject = output.Subject.LOCAL
        client = mode.Client.LOCAL
    else:
        _config = system.config[client]
        _subject = output.host_to_subject(client)

    if not 'scripts' in _config:

        return

    if f'{script}' in _config['scripts']:
        output.message(
            _subject,
            f'Running script {client}',
            True
        )
        mode.run_command(
            _config['scripts'][script],
            client
        )

def check_rsync_version():
    """
    Check rsync version
    :return:
    """
    _raw_version = mode.run_command(
        'rsync --version',
        mode.Client.LOCAL,
        True
    )
    _version = parse_version(_raw_version)
    output.message(
        output.Subject.LOCAL,
        f'rsync version {_version}'
    )


def check_sshpass_version():

    """
    Check sshpass version
    :return:
    """
    _raw_version = mode.run_command(
        'sshpass -V',
        mode.Client.LOCAL,
        force_output=True,
        allow_fail=True
    )
    _version = parse_version(_raw_version)

    if _version:
        output.message(
            output.Subject.LOCAL,
            f'sshpass version {_version}'
        )
        system.config['use_sshpass'] = True
        return True


def parse_version(output):
param = 5

class Foo:
    def __init__(self, param):   # "param" would be flagged here
        self.param = param
    """
    Parse version out of console output
    https://stackoverflow.com/a/60730346
    :param output: String
    :return:
    """
    _version_pattern = r'\d+(=?\.(\d+(=?\.(\d+)*)*)*)*'
    _regex_matcher = re.compile(_version_pattern)
    _version = _regex_matcher.search(output)
    if _version:

        return _version.group(0)
    else:
        return None


def get_file_from_path(path):
    """
    Trims a path string to retrieve the file
    :param path:
    :return: file
    """
    return path.split('/')[-1]


def confirm(prompt=None, resp=False):
    """
    https://code.activestate.com/recipes/541096-prompt-the-user-for-confirmation/

    prompts for yes or no response from the user. Returns True for yes and
    False for no.

    'resp' should be set to the default value assumed by the caller when
    user simply types ENTER.

    >>> confirm(prompt='Create Directory?', resp=True)
    Create Directory? [Y|n]:
    True
    >>> confirm(prompt='Create Directory?', resp=False)
    Create Directory? [y|N]:
    False

    """

    if prompt is None:
        prompt = 'Confirm'

    if resp:
        prompt = '%s [%s|%s]: ' % (prompt, 'Y', 'n')
    else:
        prompt = '%s [%s|%s]: ' % (prompt, 'y', 'N')

    while True:
        ans = input(prompt)
        if not ans:
            return resp
        if ans not in ['y', 'Y', 'n', 'N']:
            print('Please enter y or n.')
            continue
        if ans == 'y' or ans == 'Y':

            return True
        if ans == 'n' or ans == 'N':

            return False


1			#!/usr/bin/env python3
2			# -- coding: future_fstrings --
3
4			"""
5			Helper script
6			"""
7
8			import shutil
9			import os
10			import re
11			from db_sync_tool.utility import mode, system, output
12			from db_sync_tool.remote import utility as remote_utility
13
14
15			def clean_up():
16			"""
17			Clean up
18			:return:
19			"""
20			if not mode.is_import():
21			remote_utility.remove_target_database_dump()
22			if mode.get_sync_mode() == mode.SyncMode.PROXY:
23			remove_temporary_data_dir()
24
25
26			def remove_temporary_data_dir():
27			"""
28			Remove temporary data directory for storing database dump files
29			:return:
30			"""
31			if os.path.exists(system.default_local_sync_path):
32			output.message(
33			output.Subject.LOCAL,
34			'Cleaning up',
35			True
36			)
37			shutil.rmtree(system.default_local_sync_path)
38
39
40			def clean_up_dump_dir(client, path, num=5):
41			"""
42			Clean up the dump directory from old dump files (only affect .sql and .tar.gz files)
43			:param client:
44			:param path:
45			:param num:
46			:return:
47			"""
48			# Distinguish stat command on os system (Darwin\|Linux)
49			if check_os(client).strip() == 'Darwin':
50			_command = get_command(client, 'stat') + ' -f "%Sm %N" ' + path + ' \| ' + get_command(
51			client,
52			'sort') + ' -rn \| ' + get_command(
53			client, 'grep') + ' -E ".tar.gz\|.sql"'
			0 ignored issues – show Coding Style introduced 2021-04-13 18:13 UTC by Report Bug Copy Issue Report Wrong hanging indentation (add 4 spaces). Loading history...
54			else:
55			_command = get_command(client, 'stat') + ' -c "%y %n" ' + path + ' \| ' + \
56			get_command(client,'sort') + ' -rn \| ' + get_command(client, 'grep') + \
			0 ignored issues – show Coding Style introduced 2021-04-13 18:13 UTC by Report Bug Copy Issue Report Exactly one space required after comma Loading history...
57			' -E ".tar.gz\|.sql"'
58
59			# List files in directory sorted by change date
60			_files = mode.run_command(
61			_command,
62			client,
63			True
64			).splitlines()
65
66			for i in range(len(_files)):
			0 ignored issues – show unused-code introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Consider using enumerate instead of iterating with range and len Loading history...
67			_filename = _files[i].rsplit(' ', 1)[-1]
68
69			# Remove oldest files chosen by keep_dumps count
70			if not i < num:
71			mode.run_command(
72			'rm ' + _filename,
73			client
74			)
75
76
77			def check_os(client):
78			"""
79			Check which system is running (Linux\|Darwin)
80			:param client:
81			:return:
82			"""
83			return mode.run_command(
84			get_command(client, 'uname') + ' -s',
85			client,
86			True
87			)
88
89
90			def get_command(client, command):
91			"""
92			Get command helper for overriding default commands on the given client
93			:param client:
94			:param command:
95			:return: String command
96			"""
97			if 'console' in system.config[client]:
98			if command in system.config[client]['console']:
99			return system.config[client]['console'][command]
100			return command
101
102
103			def get_dump_dir(client):
104			"""
105			Get database dump directory by client
106			:param client:
107			:return: String path
108			"""
109			if system.config[f'default_{client}_dump_dir']:
			0 ignored issues – show unused-code introduced 2020-11-26 10:20 UTC by Report Bug Copy Issue Report Unnecessary "else" after "return" Loading history...
110			return '/tmp/'
111			else:
112			return system.config[client]['dump_dir']
113
114
115			def check_and_create_dump_dir(client, path):
116			"""
117			Check if a path exists on the client system and creates the given path if necessary
118			:param client:
119			:param path:
120			:return:
121			"""
122			mode.run_command(
123			'[ ! -d "' + path + '" ] && mkdir -p "' + path + '"',
124			client
125			)
126
127
128			def get_ssh_host_name(client, with_user=False):
129			"""
130			Format ssh host name depending on existing client name
131			:param client:
132			:param with_user:
133			:return:
134			"""
135			if not 'user' in system.config[client] and not 'host' in system.config[client]:
136			return ''
137
138			if with_user:
139			_host = system.config[client]['user'] + '@' + system.config[client]['host']
140			else:
141			_host = system.config[client]['host']
142
143			if 'name' in system.config[client]:
			0 ignored issues – show unused-code introduced 2020-11-27 11:55 UTC by Report Bug Copy Issue Report Unnecessary "else" after "return" Loading history...
144			return output.CliFormat.BOLD + system.config[client][
145			'name'] + output.CliFormat.ENDC + output.CliFormat.BLACK + ' (' + _host + ')' + \
146			output.CliFormat.ENDC
147			else:
148			return _host
149
150
151			def create_local_temporary_data_dir():
152			"""
153			Create local temporary data dir
154			:return:
155			"""
156			# @ToDo: Combine with check_and_create_dump_dir()
157			if not os.path.exists(system.default_local_sync_path):
158			os.makedirs(system.default_local_sync_path)
159
160
161			def dict_to_args(dict):
			0 ignored issues – show Bug Best Practice introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report This seems to re-define the built-in `dict`. It is generally discouraged to redefine built-ins as this makes code very hard to read. Loading history...
162			"""
163			Convert an dictionary to a args list
164			:param dict: Dictionary
165			:return: List
166			"""
167			_args = []
168			for key, val in dict.items():
169			if isinstance(val, bool):
170			if val:
171			_args.append(f'--{key}')
172			else:
173			_args.append(f'--{key}')
174			_args.append(str(val))
175			if len(_args) == 0:
176			return None
177			return _args
178
179
180			def check_file_exists(client, path):
181			"""
182			Check if a file exists
183			:param client: String
184			:param path: String file path
185			:return: Boolean
186			"""
187			return mode.run_command(f'[ -f {path} ] && echo "1"', client, True) == '1'
188
189
190			def run_script(client=None, script='before'):
191			"""
192			Executing script command
193			:param client: String
194			:param script: String
195			:return:
196			"""
197			if client is None:
198			_config = system.config
199			_subject = output.Subject.LOCAL
200			client = mode.Client.LOCAL
201			else:
202			_config = system.config[client]
203			_subject = output.host_to_subject(client)
204
205			if not 'scripts' in _config:
			0 ignored issues – show Unused Code introduced 2021-01-16 20:55 UTC by Report Bug Copy Issue Report Consider changing "not 'scripts' in _config" to "'scripts' not in _config" Loading history...
206			return
207
208			if f'{script}' in _config['scripts']:
209			output.message(
210			_subject,
211			f'Running script {client}',
212			True
213			)
214			mode.run_command(
215			_config['scripts'][script],
216			client
217			)
218
219			def check_rsync_version():
220			"""
221			Check rsync version
222			:return:
223			"""
224			_raw_version = mode.run_command(
225			'rsync --version',
226			mode.Client.LOCAL,
227			True
228			)
229			_version = parse_version(_raw_version)
230			output.message(
231			output.Subject.LOCAL,
232			f'rsync version {_version}'
233			)
234
235
236			def check_sshpass_version():
			0 ignored issues – show Unused Code introduced 2022-02-13 19:37 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...
237			"""
238			Check sshpass version
239			:return:
240			"""
241			_raw_version = mode.run_command(
242			'sshpass -V',
243			mode.Client.LOCAL,
244			force_output=True,
245			allow_fail=True
246			)
247			_version = parse_version(_raw_version)
248
249			if _version:
250			output.message(
251			output.Subject.LOCAL,
252			f'sshpass version {_version}'
253			)
254			system.config['use_sshpass'] = True
255			return True
256
257
258			def parse_version(output):
			0 ignored issues – show Comprehensibility Bug introduced 2022-02-13 19:37 UTC by Report Bug Copy Issue Report `output` is re-defining a name which is already available in the outer-scope (previously defined on line `11`). It is generally a bad practice to shadow variables from the outer-scope. In most cases, this is done unintentionally and might lead to unexpected behavior: param = 5 class Foo: def __init__(self, param): # "param" would be flagged here self.param = param Loading history...
259			"""
260			Parse version out of console output
261			https://stackoverflow.com/a/60730346
262			:param output: String
263			:return:
264			"""
265			_version_pattern = r'\d+(=?\.(\d+(=?\.(\d+))))'
266			_regex_matcher = re.compile(_version_pattern)
267			_version = _regex_matcher.search(output)
268			if _version:
			0 ignored issues – show unused-code introduced 2022-02-13 19:37 UTC by Report Bug Copy Issue Report Unnecessary "else" after "return" Loading history...
269			return _version.group(0)
270			else:
271			return None
272
273
274			def get_file_from_path(path):
275			"""
276			Trims a path string to retrieve the file
277			:param path:
278			:return: file
279			"""
280			return path.split('/')[-1]
281
282
283			def confirm(prompt=None, resp=False):
284			"""
285			https://code.activestate.com/recipes/541096-prompt-the-user-for-confirmation/
286
287			prompts for yes or no response from the user. Returns True for yes and
288			False for no.
289
290			'resp' should be set to the default value assumed by the caller when
291			user simply types ENTER.
292
293			>>> confirm(prompt='Create Directory?', resp=True)
294			Create Directory? [Y\|n]:
295			True
296			>>> confirm(prompt='Create Directory?', resp=False)
297			Create Directory? [y\|N]:
298			False
299
300			"""
301
302			if prompt is None:
303			prompt = 'Confirm'
304
305			if resp:
306			prompt = '%s [%s\|%s]: ' % (prompt, 'Y', 'n')
307			else:
308			prompt = '%s [%s\|%s]: ' % (prompt, 'y', 'N')
309
310			while True:
311			ans = input(prompt)
312			if not ans:
313			return resp
314			if ans not in ['y', 'Y', 'n', 'N']:
315			print('Please enter y or n.')
316			continue
317			if ans == 'y' or ans == 'Y':
			0 ignored issues – show Unused Code introduced 2020-12-23 09:57 UTC by Report Bug Copy Issue Report Consider merging these comparisons with "in" to "ans in ('y', 'Y')" Loading history...
318			return True
319			if ans == 'n' or ans == 'N':
			0 ignored issues – show Unused Code introduced 2020-12-23 09:57 UTC by Report Bug Copy Issue Report Consider merging these comparisons with "in" to "ans in ('n', 'N')" Loading history...
320			return False
321

jackd248 / db-sync-tool

Push — master ( 190e21...7fdd2b )

db_sync_tool.utility.helper.confirm() C

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like