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

Completed

Push — master ( 558b12...bc517a )

by Konrad

created 2020-12-23 09:45 UTC

db_sync_tool.utility.helper.confirm() C

↳ Parent: db_sync_tool.utility.helper

Complexity

Conditions

Size

Total Lines	39
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

How to fix Complexity

#!/usr/bin/env python3

# -*- coding: future_fstrings -*-

import sys

import getpass

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

#
# FUNCTIONS
#


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):
        shutil.rmtree(system.default_local_sync_path)
        output.message(
            output.Subject.LOCAL,
            'Cleaning up',
            True
        )


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 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.mkdir(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 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
class SomeClass:
    def __init__(self):
        self._x = 5

    def squared(self):
        return self._x * self._x

some_class = SomeClass()
some_class.squared()        # Flagged, as the return value is not used
print(some_class.squared()) # Ok
            '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
			0 ignored issues – show introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Missing module docstring Loading history...
2			# -- coding: future_fstrings --
3
4			import sys
			0 ignored issues – show Unused Code introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report The import `sys` seems to be unused. Loading history...
5			import getpass
			0 ignored issues – show Unused Code introduced 2020-11-26 10:20 UTC by Report Bug Copy Issue Report The import `getpass` seems to be unused. Loading history...
6			import shutil
7			import os
8			from db_sync_tool.utility import mode, system, output
9			from db_sync_tool.remote import utility as remote_utility
10
11			#
12			# FUNCTIONS
13			#
14
15
16			def clean_up():
17			"""
18			Clean up
19			:return:
20			"""
21			if not mode.is_import():
22			remote_utility.remove_target_database_dump()
23			if mode.get_sync_mode() == mode.SyncMode.PROXY:
24			remove_temporary_data_dir()
25
26
27			def remove_temporary_data_dir():
28			"""
29			Remove temporary data directory for storing database dump files
30			:return:
31			"""
32			if os.path.exists(system.default_local_sync_path):
33			shutil.rmtree(system.default_local_sync_path)
34			output.message(
35			output.Subject.LOCAL,
36			'Cleaning up',
37			True
38			)
39
40
41			def clean_up_dump_dir(client, path, num=5):
42			"""
43			Clean up the dump directory from old dump files (only affect .sql and .tar.gz files)
44			:param client:
45			:param path:
46			:param num:
47			:return:
48			"""
49			# Distinguish stat command on os system (Darwin\|Linux)
50			if check_os(client).strip() == 'Darwin':
51			_command = get_command(client, 'stat') + ' -f "%Sm %N" ' + path + ' \| ' + get_command(client,
			0 ignored issues – show Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (101/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
52			'sort') + ' -rn \| ' + get_command(
			0 ignored issues – show Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (128/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
53			client, 'grep') + ' -E ".tar.gz\|.sql"'
			0 ignored issues – show Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Wrong hanging indentation (add 86 spaces). Loading history...
54			else:
55			_command = get_command(client, 'stat') + ' -c "%y %n" ' + path + ' \| ' + get_command(client,
56			'sort') + ' -rn \| ' + get_command(
			0 ignored issues – show Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (127/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
57			client, 'grep') + ' -E ".tar.gz\|.sql"'
			0 ignored issues – show Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Wrong hanging indentation (add 85 spaces). Loading history...
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 with_user:
136			_host = system.config[client]['user'] + '@' + system.config[client]['host']
137			else:
138			_host = system.config[client]['host']
139
140			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...
141			return output.CliFormat.BOLD + system.config[client][
142			'name'] + output.CliFormat.ENDC + output.CliFormat.BLACK + ' (' + _host + ')' + output.CliFormat.ENDC
			0 ignored issues – show Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (113/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
143			else:
144			return _host
145
146
147			def create_local_temporary_data_dir():
148			"""
149			Create local temporary data dir
150			:return:
151			"""
152			# @ToDo: Combine with check_and_create_dump_dir()
153			if not os.path.exists(system.default_local_sync_path):
154			os.mkdir(system.default_local_sync_path)
155
156
157			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...
158			"""
159			Convert an dictionary to a args list
160			:param dict: Dictionary
161			:return: List
162			"""
163			_args = []
164			for key, val in dict.items():
165			if isinstance(val, bool):
166			if val:
167			_args.append(f'--{key}')
168			else:
169			_args.append(f'--{key}')
170			_args.append(str(val))
171			if len(_args) == 0:
172			return None
173			return _args
174
175
176			def check_file_exists(client, path):
177			"""
178			Check if a file exists
179			:param client: String
180			:param path: String file path
181			:return: Boolean
182			"""
183			return mode.run_command(f'[ -f {path} ] && echo "1"', client, True) == '1'
184
185
186			def confirm(prompt=None, resp=False):
187			"""
188			https://code.activestate.com/recipes/541096-prompt-the-user-for-confirmation/
189
190			prompts for yes or no response from the user. Returns True for yes and
191			False for no.
192
193			'resp' should be set to the default value assumed by the caller when
194			user simply types ENTER.
195
196			>>> confirm(prompt='Create Directory?', resp=True)
197			Create Directory? [Y\|n]:
198			True
199			>>> confirm(prompt='Create Directory?', resp=False)
200			Create Directory? [y\|N]:
201			False
202
203			"""
204
205			if prompt is None:
206			prompt = 'Confirm'
207
208			if resp:
209			prompt = '%s [%s\|%s]: ' % (prompt, 'Y', 'n')
210			else:
211			prompt = '%s [%s\|%s]: ' % (prompt, 'y', 'N')
212
213			while True:
214			ans = input(prompt)
215			if not ans:
216			return resp
217			if ans not in ['y', 'Y', 'n', 'N']:
218			print
			0 ignored issues – show Unused Code introduced 2020-12-23 09:57 UTC by Report Bug Copy Issue Report This statement seems to have no effect and could be removed. This issue is typically triggered when a function that does not have side-effects is called and the return value is discarded: class SomeClass: def __init__(self): self._x = 5 def squared(self): return self._x * self._x some_class = SomeClass() some_class.squared() # Flagged, as the return value is not used print(some_class.squared()) # Ok Loading history...
219			'please enter y or n.'
			0 ignored issues – show Unused Code introduced 2020-12-23 09:57 UTC by Report Bug Copy Issue Report This string statement has no effect and could be removed. Loading history...
220			continue
221			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...
222			return True
223			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...
224			return False
225

jackd248 / db-sync-tool

Push — master ( 558b12...bc517a )

db_sync_tool.utility.helper.confirm() C

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like