db_sync_tool.utility.helper.clean_up_dump_dir() - Code Metrics - Inspection of "[CleanUp] pylint" - jackd248/db-sync-tool - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( e2438a...6bbf05 )

by Konrad

created 2021-04-13 18:11 UTC

db_sync_tool.utility.helper.clean_up_dump_dir() A

↳ Parent: db_sync_tool.utility.helper

Complexity

Conditions

Size

Total Lines	34
Code Lines	20

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	20
dl	0
loc	34
rs	9.4
c	0
b	0
f	0
cc	4
nop	3

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

"""
Helper script
"""

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

jackd248 / db-sync-tool

Push — master ( e2438a...6bbf05 )

db_sync_tool.utility.helper.clean_up_dump_dir() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like