db_sync_tool.utility.helper.run_script() - Code Metrics - Inspection of "[Task] Extend scripts feature" - jackd248/db-sync-tool - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 2a74fd...6305c6 )

by Konrad

created 2021-01-16 20:54 UTC

db_sync_tool.utility.helper.run_script() A

↳ Parent: db_sync_tool.utility.helper

Complexity

Conditions

Size

Total Lines	27
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	17
dl	0
loc	27
rs	9.55
c	0
b	0
f	0
cc	4
nop	2

#!/usr/bin/env python3

# -*- coding: future_fstrings -*-

import shutil
import os
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):
        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 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.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 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 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
			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 shutil
5			import os
6			from db_sync_tool.utility import mode, system, output
7			from db_sync_tool.remote import utility as remote_utility
8
9
10			def clean_up():
11			"""
12			Clean up
13			:return:
14			"""
15			if not mode.is_import():
16			remote_utility.remove_target_database_dump()
17			if mode.get_sync_mode() == mode.SyncMode.PROXY:
18			remove_temporary_data_dir()
19
20
21			def remove_temporary_data_dir():
22			"""
23			Remove temporary data directory for storing database dump files
24			:return:
25			"""
26			if os.path.exists(system.default_local_sync_path):
27			shutil.rmtree(system.default_local_sync_path)
28			output.message(
29			output.Subject.LOCAL,
30			'Cleaning up',
31			True
32			)
33
34
35			def clean_up_dump_dir(client, path, num=5):
36			"""
37			Clean up the dump directory from old dump files (only affect .sql and .tar.gz files)
38			:param client:
39			:param path:
40			:param num:
41			:return:
42			"""
43			# Distinguish stat command on os system (Darwin\|Linux)
44			if check_os(client).strip() == 'Darwin':
45			_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...
46			'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...
47			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...
48			else:
49			_command = get_command(client, 'stat') + ' -c "%y %n" ' + path + ' \| ' + get_command(client,
50			'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...
51			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...
52
53			# List files in directory sorted by change date
54			_files = mode.run_command(
55			_command,
56			client,
57			True
58			).splitlines()
59
60			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...
61			_filename = _files[i].rsplit(' ', 1)[-1]
62
63			# Remove oldest files chosen by keep_dumps count
64			if not i < num:
65			mode.run_command(
66			'rm ' + _filename,
67			client
68			)
69
70
71			def check_os(client):
72			"""
73			Check which system is running (Linux\|Darwin)
74			:param client:
75			:return:
76			"""
77			return mode.run_command(
78			get_command(client, 'uname') + ' -s',
79			client,
80			True
81			)
82
83
84			def get_command(client, command):
85			"""
86			Get command helper for overriding default commands on the given client
87			:param client:
88			:param command:
89			:return: String command
90			"""
91			if 'console' in system.config[client]:
92			if command in system.config[client]['console']:
93			return system.config[client]['console'][command]
94			return command
95
96
97			def get_dump_dir(client):
98			"""
99			Get database dump directory by client
100			:param client:
101			:return: String path
102			"""
103			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...
104			return '/tmp/'
105			else:
106			return system.config[client]['dump_dir']
107
108
109			def check_and_create_dump_dir(client, path):
110			"""
111			Check if a path exists on the client system and creates the given path if necessary
112			:param client:
113			:param path:
114			:return:
115			"""
116			mode.run_command(
117			'[ ! -d "' + path + '" ] && mkdir -p "' + path + '"',
118			client
119			)
120
121
122			def get_ssh_host_name(client, with_user=False):
123			"""
124			Format ssh host name depending on existing client name
125			:param client:
126			:param with_user:
127			:return:
128			"""
129			if not 'user' in system.config[client] and not 'host' in system.config[client]:
130			return ''
131
132			if with_user:
133			_host = system.config[client]['user'] + '@' + system.config[client]['host']
134			else:
135			_host = system.config[client]['host']
136
137			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...
138			return output.CliFormat.BOLD + system.config[client][
139			'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...
140			else:
141			return _host
142
143
144			def create_local_temporary_data_dir():
145			"""
146			Create local temporary data dir
147			:return:
148			"""
149			# @ToDo: Combine with check_and_create_dump_dir()
150			if not os.path.exists(system.default_local_sync_path):
151			os.mkdir(system.default_local_sync_path)
152
153
154			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...
155			"""
156			Convert an dictionary to a args list
157			:param dict: Dictionary
158			:return: List
159			"""
160			_args = []
161			for key, val in dict.items():
162			if isinstance(val, bool):
163			if val:
164			_args.append(f'--{key}')
165			else:
166			_args.append(f'--{key}')
167			_args.append(str(val))
168			if len(_args) == 0:
169			return None
170			return _args
171
172
173			def check_file_exists(client, path):
174			"""
175			Check if a file exists
176			:param client: String
177			:param path: String file path
178			:return: Boolean
179			"""
180			return mode.run_command(f'[ -f {path} ] && echo "1"', client, True) == '1'
181
182
183			def run_script(client=None, script='before'):
184			"""
185			Executing script command
186			:param client: String
187			:param script: String
188			:return:
189			"""
190			if client is None:
191			_config = system.config
192			_subject = output.Subject.LOCAL
193			client = mode.Client.LOCAL
194			else:
195			_config = system.config[client]
196			_subject = output.host_to_subject(client)
197
198			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...
199			return
200
201			if f'{script}' in _config['scripts']:
202			output.message(
203			_subject,
204			f'Running script {client}',
205			True
206			)
207			mode.run_command(
208			_config['scripts'][script],
209			client
210			)
211
212
213			def confirm(prompt=None, resp=False):
214			"""
215			https://code.activestate.com/recipes/541096-prompt-the-user-for-confirmation/
216
217			prompts for yes or no response from the user. Returns True for yes and
218			False for no.
219
220			'resp' should be set to the default value assumed by the caller when
221			user simply types ENTER.
222
223			>>> confirm(prompt='Create Directory?', resp=True)
224			Create Directory? [Y\|n]:
225			True
226			>>> confirm(prompt='Create Directory?', resp=False)
227			Create Directory? [y\|N]:
228			False
229
230			"""
231
232			if prompt is None:
233			prompt = 'Confirm'
234
235			if resp:
236			prompt = '%s [%s\|%s]: ' % (prompt, 'Y', 'n')
237			else:
238			prompt = '%s [%s\|%s]: ' % (prompt, 'y', 'N')
239
240			while True:
241			ans = input(prompt)
242			if not ans:
243			return resp
244			if ans not in ['y', 'Y', 'n', 'N']:
245			print('Please enter y or n.')
246			continue
247			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...
248			return True
249			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...
250			return False
251

jackd248 / db-sync-tool

Push — master ( 2a74fd...6305c6 )

db_sync_tool.utility.helper.run_script() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like