db_sync_tool.utility.helper.remove_temporary_data_dir() - Code Metrics - Inspection of "release: version 2.11.10" - jackd248/db-sync-tool - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( c87e81...fcabd8 )

by Konrad

created 2025-04-09 15:25 UTC

remove_temporary_data_dir() A

↳ Parent: db_sync_tool.utility.helper

Complexity

Conditions

Size

Total Lines	12
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	7
nop	0
dl	0
loc	12
rs	10
c	0
b	0
f	0

#!/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, minimal=False):
    """
    Format ssh host name depending on existing client name
    :param client:
    :param with_user:
    :param short:
    :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]:
        if minimal:

            return system.config[client]['name']
        else:
            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


def clean_db_config(config):
    """
    Iterates over all entries of a dictionary and removes enclosing inverted commas
    from the values, if present.

    :param config: The dictionary to be edited
    :return: A new dictionary with adjusted values
    """
    # Iterate over all entries in the dictionary and use the inverted comma function
    return {key: remove_surrounding_quotes(value) for key, value in config.items()}


def remove_surrounding_quotes(s):

    """
    Removes the enclosing inverted commas (single or double),
    if there are quotes at both the beginning and the end of the string.

    :param s: The string to be checked
    :return: The string without enclosing quotes, if available
    """
    if isinstance(s, str):
        if s.startswith('"') and s.endswith('"'):

            return s[1:-1]
        elif s.startswith("'") and s.endswith("'"):
            return s[1:-1]
    return s


def run_sed_command(client, command):
    """
    Executes a sed command on the specified client, trying -E first and falling back to -r if -E fails.


    :param client: The client on which the sed command should be executed.
    :param command: The sed command to execute (excluding the sed options).
    :return: The result of the sed command as a cleaned string (with newlines removed).
    """
    # Check if the client supports -E or -r option for sed
    option = mode.run_command(
         f"echo | {get_command(client, 'sed')} -E '' >/dev/null 2>&1 && echo -E || (echo | {get_command(client, 'sed')} -r '' >/dev/null 2>&1 && echo -r)",

         client,

         True

    )
    # If neither option is supported, default to -E
    if option == '':
        option = '-E'

    return mode.run_command(
        f"{get_command(client, 'sed')} -n {option} {command}",
        client,
        True
    ).strip().replace('\n', '')


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, minimal=False):
129			"""
130			Format ssh host name depending on existing client name
131			:param client:
132			:param with_user:
133			:param short:
134			:return:
135			"""
136			if not 'user' in system.config[client] and not 'host' in system.config[client]:
137			return ''
138
139			if with_user:
140			_host = system.config[client]['user'] + '@' + system.config[client]['host']
141			else:
142			_host = system.config[client]['host']
143
144			if 'name' in system.config[client]:
145			if minimal:
			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...
146			return system.config[client]['name']
147			else:
148			return output.CliFormat.BOLD + system.config[client][
149			'name'] + output.CliFormat.ENDC + output.CliFormat.BLACK + ' (' + _host + ')' + \
150			output.CliFormat.ENDC
151			else:
152			return _host
153
154
155			def create_local_temporary_data_dir():
156			"""
157			Create local temporary data dir
158			:return:
159			"""
160			# @ToDo: Combine with check_and_create_dump_dir()
161			if not os.path.exists(system.default_local_sync_path):
162			os.makedirs(system.default_local_sync_path)
163
164
165			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...
166			"""
167			Convert an dictionary to a args list
168			:param dict: Dictionary
169			:return: List
170			"""
171			_args = []
172			for key, val in dict.items():
173			if isinstance(val, bool):
174			if val:
175			_args.append(f'--{key}')
176			else:
177			_args.append(f'--{key}')
178			_args.append(str(val))
179			if len(_args) == 0:
180			return None
181			return _args
182
183
184			def check_file_exists(client, path):
185			"""
186			Check if a file exists
187			:param client: String
188			:param path: String file path
189			:return: Boolean
190			"""
191			return mode.run_command(f'[ -f {path} ] && echo "1"', client, True) == '1'
192
193
194			def run_script(client=None, script='before'):
195			"""
196			Executing script command
197			:param client: String
198			:param script: String
199			:return:
200			"""
201			if client is None:
202			_config = system.config
203			_subject = output.Subject.LOCAL
204			client = mode.Client.LOCAL
205			else:
206			_config = system.config[client]
207			_subject = output.host_to_subject(client)
208
209			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...
210			return
211
212			if f'{script}' in _config['scripts']:
213			output.message(
214			_subject,
215			f'Running script {client}',
216			True
217			)
218			mode.run_command(
219			_config['scripts'][script],
220			client
221			)
222
223
224			def check_rsync_version():
225			"""
226			Check rsync version
227			:return:
228			"""
229			_raw_version = mode.run_command(
230			'rsync --version',
231			mode.Client.LOCAL,
232			True
233			)
234			_version = parse_version(_raw_version)
235			output.message(
236			output.Subject.LOCAL,
237			f'rsync version {_version}'
238			)
239
240
241			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...
242			"""
243			Check sshpass version
244			:return:
245			"""
246			_raw_version = mode.run_command(
247			'sshpass -V',
248			mode.Client.LOCAL,
249			force_output=True,
250			allow_fail=True
251			)
252			_version = parse_version(_raw_version)
253
254			if _version:
255			output.message(
256			output.Subject.LOCAL,
257			f'sshpass version {_version}'
258			)
259			system.config['use_sshpass'] = True
260			return True
261
262
263			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...
264			"""
265			Parse version out of console output
266			https://stackoverflow.com/a/60730346
267			:param output: String
268			:return:
269			"""
270			_version_pattern = r'\d+(=?\.(\d+(=?\.(\d+))))'
271			_regex_matcher = re.compile(_version_pattern)
272			_version = _regex_matcher.search(output)
273			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...
274			return _version.group(0)
275			else:
276			return None
277
278
279			def get_file_from_path(path):
280			"""
281			Trims a path string to retrieve the file
282			:param path:
283			:return: file
284			"""
285			return path.split('/')[-1]
286
287
288			def confirm(prompt=None, resp=False):
289			"""
290			https://code.activestate.com/recipes/541096-prompt-the-user-for-confirmation/
291
292			prompts for yes or no response from the user. Returns True for yes and
293			False for no.
294
295			'resp' should be set to the default value assumed by the caller when
296			user simply types ENTER.
297
298			>>> confirm(prompt='Create Directory?', resp=True)
299			Create Directory? [Y\|n]:
300			True
301			>>> confirm(prompt='Create Directory?', resp=False)
302			Create Directory? [y\|N]:
303			False
304
305			"""
306
307			if prompt is None:
308			prompt = 'Confirm'
309
310			if resp:
311			prompt = '%s [%s\|%s]: ' % (prompt, 'Y', 'n')
312			else:
313			prompt = '%s [%s\|%s]: ' % (prompt, 'y', 'N')
314
315			while True:
316			ans = input(prompt)
317			if not ans:
318			return resp
319			if ans not in ['y', 'Y', 'n', 'N']:
320			print('Please enter y or n.')
321			continue
322			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...
323			return True
324			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...
325			return False
326
327
328			def clean_db_config(config):
329			"""
330			Iterates over all entries of a dictionary and removes enclosing inverted commas
331			from the values, if present.
332
333			:param config: The dictionary to be edited
334			:return: A new dictionary with adjusted values
335			"""
336			# Iterate over all entries in the dictionary and use the inverted comma function
337			return {key: remove_surrounding_quotes(value) for key, value in config.items()}
338
339
340			def remove_surrounding_quotes(s):
			0 ignored issues – show Coding Style Naming introduced 2024-09-23 08:56 UTC by Report Bug Copy Issue Report Argument name "s" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\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...
341			"""
342			Removes the enclosing inverted commas (single or double),
343			if there are quotes at both the beginning and the end of the string.
344
345			:param s: The string to be checked
346			:return: The string without enclosing quotes, if available
347			"""
348			if isinstance(s, str):
349			if s.startswith('"') and s.endswith('"'):
			0 ignored issues – show unused-code introduced 2024-09-23 08:56 UTC by Report Bug Copy Issue Report Unnecessary "elif" after "return" Loading history...
350			return s[1:-1]
351			elif s.startswith("'") and s.endswith("'"):
352			return s[1:-1]
353			return s
354
355
356			def run_sed_command(client, command):
357			"""
358			Executes a sed command on the specified client, trying -E first and falling back to -r if -E fails.
			0 ignored issues – show Coding Style introduced 2025-04-09 15:26 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (103/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
359
360			:param client: The client on which the sed command should be executed.
361			:param command: The sed command to execute (excluding the sed options).
362			:return: The result of the sed command as a cleaned string (with newlines removed).
363			"""
364			# Check if the client supports -E or -r option for sed
365			option = mode.run_command(
366			f"echo \| {get_command(client, 'sed')} -E '' >/dev/null 2>&1 && echo -E \|\| (echo \| {get_command(client, 'sed')} -r '' >/dev/null 2>&1 && echo -r)",
			0 ignored issues – show Coding Style introduced 2025-04-09 15:26 UTC by Report Bug Copy Issue Report Wrong hanging indentation (remove 1 space). Loading history... Coding Style introduced 2025-04-09 15:26 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (155/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
367			client,
			0 ignored issues – show Coding Style introduced 2025-04-09 15:26 UTC by Report Bug Copy Issue Report Wrong hanging indentation (remove 1 space). Loading history...
368			True
			0 ignored issues – show Coding Style introduced 2025-04-09 15:26 UTC by Report Bug Copy Issue Report Wrong hanging indentation (remove 1 space). Loading history...
369			)
370			# If neither option is supported, default to -E
371			if option == '':
372			option = '-E'
373
374			return mode.run_command(
375			f"{get_command(client, 'sed')} -n {option} {command}",
376			client,
377			True
378			).strip().replace('\n', '')
379

jackd248 / db-sync-tool

Push — master ( c87e81...fcabd8 )

remove_temporary_data_dir() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like