db_sync_tool.utility.mode.run_command() - Code Metrics - jackd248/db-sync-tool - Measure and Improve Code Quality continuously with Scrutinizer

db_sync_tool.utility.mode.run_command() C
last analyzed 2025-07-23 08:59 UTC

↳ Parent: db_sync_tool.utility.mode

Complexity

Conditions

Size

Total Lines	36
Code Lines	19

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	19
dl	0
loc	36
rs	5.9999
c	0
b	0
f	0
cc	10
nop	5

How to fix Complexity

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

"""
Mode script
"""

import subprocess
import sys

from db_sync_tool.utility import system, output, helper
from db_sync_tool.remote import system as remote_system


#
# GLOBALS
#

class Client:

    ORIGIN = 'origin'
    TARGET = 'target'
    LOCAL = 'local'


class SyncMode:
    """
    Sync Mode
    """

    DUMP_LOCAL = 'DUMP_LOCAL'
    DUMP_REMOTE = 'DUMP_REMOTE'
    IMPORT_LOCAL = 'IMPORT_LOCAL'
    IMPORT_REMOTE = 'IMPORT_REMOTE'
    RECEIVER = 'RECEIVER'
    SENDER = 'SENDER'
    PROXY = 'PROXY'
    SYNC_REMOTE = 'SYNC_REMOTE'
    SYNC_LOCAL = 'SYNC_LOCAL'

    @staticmethod
    def is_dump_local():
        """

        :return: boolean
        """
        return SyncMode.is_full_local() and SyncMode.is_same_host() and not SyncMode.is_sync_local()

    @staticmethod
    def is_dump_remote():
        """

        :return: boolean
        """
        return SyncMode.is_full_remote() and SyncMode.is_same_host() and \
               not SyncMode.is_sync_remote()

    @staticmethod
    def is_receiver():
        """

        :return: boolean
        """
        return 'host' in system.config[Client.ORIGIN] and not SyncMode.is_proxy() and \
               not SyncMode.is_sync_remote()

    @staticmethod
    def is_sender():
        """

        :return: boolean
        """
        return 'host' in system.config[Client.TARGET] and not SyncMode.is_proxy() and \
               not SyncMode.is_sync_remote()

    @staticmethod
    def is_proxy():
        """

        :return: boolean
        """
        return SyncMode.is_full_remote()

    @staticmethod
    def is_import_local():
        """

        :return: boolean
        """
        return system.config['import'] != '' and 'host' not in system.config[Client.TARGET]

    @staticmethod
    def is_import_remote():
        """

        :return: boolean
        """
        return system.config['import'] != '' and 'host' in system.config[Client.TARGET]

    @staticmethod
    def is_sync_local():
        """

        :return: boolean
        """
        return SyncMode.is_full_local() and SyncMode.is_same_host() and SyncMode.is_same_sync()

    @staticmethod
    def is_sync_remote():
        """

        :return: boolean
        """
        return SyncMode.is_full_remote() and SyncMode.is_same_host() and SyncMode.is_same_sync()

    @staticmethod
    def is_same_sync():
        """

        :return: boolean
        """
        return ((SyncMode.is_available_configuration('path') and
                 not SyncMode.is_same_configuration('path')) or
               (SyncMode.is_available_configuration('db') and

                not SyncMode.is_same_configuration('db')))

    @staticmethod
    def is_full_remote():
        """

        :return: boolean
        """
        return SyncMode.is_available_configuration('host')

    @staticmethod
    def is_full_local():
        """

        :return: boolean
        """
        return SyncMode.is_unavailable_configuration('host')

    @staticmethod
    def is_same_host():
        """

        :return: boolean
        """
        return SyncMode.is_same_configuration('host') and SyncMode.is_same_configuration('port') and SyncMode.is_same_configuration('user')


    @staticmethod
    def is_available_configuration(key):
        """

        :return: boolean
        """
        return key in system.config[Client.ORIGIN] and key in system.config[Client.TARGET]

    @staticmethod
    def is_unavailable_configuration(key):
        """

        :return: boolean
        """
        return key not in system.config[Client.ORIGIN] and key not in system.config[Client.TARGET]

    @staticmethod
    def is_same_configuration(key):
        """

        :return: boolean
        """
        return (SyncMode.is_available_configuration(key) and

               system.config[Client.ORIGIN][key] == system.config[Client.TARGET][key]) or \

               SyncMode.is_unavailable_configuration(key)


# Default sync mode
sync_mode = SyncMode.RECEIVER



#
# FUNCTIONS
#
def get_sync_mode():
    """
    Returning the sync mode
    :return: String sync_mode
    """
    return sync_mode


def check_sync_mode():
    """
    Checking the sync_mode based on the given configuration
    :return: String subject
    """
    global sync_mode

    _description = ''

    _modes = {
        SyncMode.RECEIVER: '(REMOTE ➔ LOCAL)',
        SyncMode.SENDER: '(LOCAL ➔ REMOTE)',
        SyncMode.PROXY: '(REMOTE ➔ LOCAL ➔ REMOTE)',
        SyncMode.DUMP_LOCAL: '(LOCAL, ONLY EXPORT)',
        SyncMode.DUMP_REMOTE: '(REMOTE, ONLY EXPORT)',
        SyncMode.IMPORT_LOCAL: '(REMOTE, ONLY IMPORT)',
        SyncMode.IMPORT_REMOTE: '(LOCAL, ONLY IMPORT)',
        SyncMode.SYNC_LOCAL: '(LOCAL ➔ LOCAL)',
        SyncMode.SYNC_REMOTE: '(REMOTE ➔ REMOTE)'
    }

    for _mode, _desc in _modes.items():
        if getattr(SyncMode, 'is_' + _mode.lower())():
            sync_mode = _mode
            _description = _desc

    if is_import():
        output.message(
            output.Subject.INFO,
            f'Import file {output.CliFormat.BLACK}{system.config["import"]}{output.CliFormat.ENDC}',
            True
        )

    system.config['is_same_client'] = SyncMode.is_same_host()

    output.message(
        output.Subject.INFO,
        f'Sync mode: {sync_mode} {output.CliFormat.BLACK}{_description}{output.CliFormat.ENDC}',
        True
    )

    check_for_protection()


def is_remote(client):
    """
    Check if given client is remote client
    :param client: String
    :return: Boolean
    """
    if client == Client.ORIGIN:

        return is_origin_remote()
    elif client == Client.TARGET:
        return is_target_remote()
    elif client == Client.LOCAL:
        return False
    else:
        return False


def is_target_remote():
    """
    Check if target is remote client
    :return: Boolean
    """
    return sync_mode in (SyncMode.SENDER, SyncMode.PROXY, SyncMode.DUMP_REMOTE,
                         SyncMode.IMPORT_REMOTE, SyncMode.SYNC_REMOTE)


def is_origin_remote():
    """
    Check if origin is remote client
    :return: Boolean
    """
    return sync_mode in (SyncMode.RECEIVER, SyncMode.PROXY, SyncMode.DUMP_REMOTE,
                         SyncMode.IMPORT_REMOTE, SyncMode.SYNC_REMOTE)


def is_import():
    """
    Check if sync mode is import
    :return: Boolean
    """
    return sync_mode in (SyncMode.IMPORT_LOCAL, SyncMode.IMPORT_REMOTE)


def is_dump():
    """
    Check if sync mode is import
    :return: Boolean
    """
    return sync_mode in (SyncMode.DUMP_LOCAL, SyncMode.DUMP_REMOTE)


def run_command(command, client, force_output=False, allow_fail=False, skip_dry_run=False):

    """
    Run command depending on the given client
    :param command: String
    :param client: String
    :param force_output: Boolean
    :param allow_fail: Boolean
    :param skip_dry_run: Boolean
    :return:
    """
    if system.config['verbose']:
        output.message(
            output.host_to_subject(client),
            output.CliFormat.BLACK + command + output.CliFormat.ENDC,
            debug=True
        )

    if system.config['dry_run'] and skip_dry_run:
        return

    if is_remote(client):
        if force_output:

            return ''.join(remote_system.run_ssh_command_by_client(client, command).readlines()).strip()

        else:
            return remote_system.run_ssh_command_by_client(client, command)
    else:
        res = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
        # Wait for the process end and print error in case of failure
        out, err = res.communicate()

        if res.wait() != 0 and err.decode() != '' and not allow_fail:
            helper.run_script(script='error')
            sys.exit(output.message(output.Subject.ERROR, err.decode(), False))

        if force_output:
            return out.decode().strip()


def check_for_protection():
    """
    Check if the target system is protected
    :return: Boolean
    """
    if sync_mode in (SyncMode.RECEIVER, SyncMode.SENDER, SyncMode.PROXY, SyncMode.SYNC_LOCAL,
                     SyncMode.SYNC_REMOTE, SyncMode.IMPORT_LOCAL, SyncMode.IMPORT_REMOTE) and \
            'protect' in system.config[Client.TARGET]:
        _host = helper.get_ssh_host_name(Client.TARGET)
        sys.exit(output.message(output.Subject.ERROR,
                                f'The host {_host} is protected against the import of a database dump. Please '

                                'check synchronisation target or adjust the host configuration.', False))





1			#!/usr/bin/env python3
2			# -- coding: future_fstrings --
3
4			"""
5			Mode script
6			"""
7
8			import subprocess
9			import sys
10
11			from db_sync_tool.utility import system, output, helper
12			from db_sync_tool.remote import system as remote_system
13
14
15			#
16			# GLOBALS
17			#
18
19			class Client:
			0 ignored issues – show introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Missing class docstring Loading history...
20			ORIGIN = 'origin'
21			TARGET = 'target'
22			LOCAL = 'local'
23
24
25			class SyncMode:
26			"""
27			Sync Mode
28			"""
29
30			DUMP_LOCAL = 'DUMP_LOCAL'
31			DUMP_REMOTE = 'DUMP_REMOTE'
32			IMPORT_LOCAL = 'IMPORT_LOCAL'
33			IMPORT_REMOTE = 'IMPORT_REMOTE'
34			RECEIVER = 'RECEIVER'
35			SENDER = 'SENDER'
36			PROXY = 'PROXY'
37			SYNC_REMOTE = 'SYNC_REMOTE'
38			SYNC_LOCAL = 'SYNC_LOCAL'
39
40			@staticmethod
41			def is_dump_local():
42			"""
43
44			:return: boolean
45			"""
46			return SyncMode.is_full_local() and SyncMode.is_same_host() and not SyncMode.is_sync_local()
47
48			@staticmethod
49			def is_dump_remote():
50			"""
51
52			:return: boolean
53			"""
54			return SyncMode.is_full_remote() and SyncMode.is_same_host() and \
55			not SyncMode.is_sync_remote()
56
57			@staticmethod
58			def is_receiver():
59			"""
60
61			:return: boolean
62			"""
63			return 'host' in system.config[Client.ORIGIN] and not SyncMode.is_proxy() and \
64			not SyncMode.is_sync_remote()
65
66			@staticmethod
67			def is_sender():
68			"""
69
70			:return: boolean
71			"""
72			return 'host' in system.config[Client.TARGET] and not SyncMode.is_proxy() and \
73			not SyncMode.is_sync_remote()
74
75			@staticmethod
76			def is_proxy():
77			"""
78
79			:return: boolean
80			"""
81			return SyncMode.is_full_remote()
82
83			@staticmethod
84			def is_import_local():
85			"""
86
87			:return: boolean
88			"""
89			return system.config['import'] != '' and 'host' not in system.config[Client.TARGET]
90
91			@staticmethod
92			def is_import_remote():
93			"""
94
95			:return: boolean
96			"""
97			return system.config['import'] != '' and 'host' in system.config[Client.TARGET]
98
99			@staticmethod
100			def is_sync_local():
101			"""
102
103			:return: boolean
104			"""
105			return SyncMode.is_full_local() and SyncMode.is_same_host() and SyncMode.is_same_sync()
106
107			@staticmethod
108			def is_sync_remote():
109			"""
110
111			:return: boolean
112			"""
113			return SyncMode.is_full_remote() and SyncMode.is_same_host() and SyncMode.is_same_sync()
114
115			@staticmethod
116			def is_same_sync():
117			"""
118
119			:return: boolean
120			"""
121			return ((SyncMode.is_available_configuration('path') and
122			not SyncMode.is_same_configuration('path')) or
123			(SyncMode.is_available_configuration('db') and
			0 ignored issues – show Coding Style introduced 2021-03-07 12:04 UTC by Report Bug Copy Issue Report Wrong continued indentation (add 1 space). Loading history...
124			not SyncMode.is_same_configuration('db')))
125
126			@staticmethod
127			def is_full_remote():
128			"""
129
130			:return: boolean
131			"""
132			return SyncMode.is_available_configuration('host')
133
134			@staticmethod
135			def is_full_local():
136			"""
137
138			:return: boolean
139			"""
140			return SyncMode.is_unavailable_configuration('host')
141
142			@staticmethod
143			def is_same_host():
144			"""
145
146			:return: boolean
147			"""
148			return SyncMode.is_same_configuration('host') and SyncMode.is_same_configuration('port') and SyncMode.is_same_configuration('user')
			0 ignored issues – show Coding Style introduced 2025-07-23 08:57 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (139/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
149
150			@staticmethod
151			def is_available_configuration(key):
152			"""
153
154			:return: boolean
155			"""
156			return key in system.config[Client.ORIGIN] and key in system.config[Client.TARGET]
157
158			@staticmethod
159			def is_unavailable_configuration(key):
160			"""
161
162			:return: boolean
163			"""
164			return key not in system.config[Client.ORIGIN] and key not in system.config[Client.TARGET]
165
166			@staticmethod
167			def is_same_configuration(key):
168			"""
169
170			:return: boolean
171			"""
172			return (SyncMode.is_available_configuration(key) and
			0 ignored issues – show Unused Code introduced 2021-03-07 12:04 UTC by Report Bug Copy Issue Report Consider using ternary (system.config[Client.ORIGIN][key] == system.config[Client.TARGET][key] if SyncMode.is_available_configuration(key) else SyncMode.is_unavailable_configuration(key)) Loading history...
173			system.config[Client.ORIGIN][key] == system.config[Client.TARGET][key]) or \
			0 ignored issues – show Coding Style introduced 2021-03-07 12:04 UTC by Report Bug Copy Issue Report Wrong continued indentation (add 1 space). Loading history...
174			SyncMode.is_unavailable_configuration(key)
175
176
177			# Default sync mode
178			sync_mode = SyncMode.RECEIVER
			0 ignored issues – show Coding Style Naming introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Constant name "sync_mode" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\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...
179
180
181			#
182			# FUNCTIONS
183			#
184			def get_sync_mode():
185			"""
186			Returning the sync mode
187			:return: String sync_mode
188			"""
189			return sync_mode
190
191
192			def check_sync_mode():
193			"""
194			Checking the sync_mode based on the given configuration
195			:return: String subject
196			"""
197			global sync_mode
			0 ignored issues – show Coding Style introduced 2020-12-28 13:48 UTC by Report Bug Copy Issue Report Usage of the `global` statement should be avoided. Usage of `global` can make code hard to read and test, its usage is generally not recommended unless you are dealing with legacy code. Loading history... Coding Style Naming introduced 2020-12-28 13:48 UTC by Report Bug Copy Issue Report Constant name "sync_mode" doesn't conform to UPPER_CASE naming style ('([^\\W\\da-z][^\\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...
198			_description = ''
199
200			_modes = {
201			SyncMode.RECEIVER: '(REMOTE ➔ LOCAL)',
202			SyncMode.SENDER: '(LOCAL ➔ REMOTE)',
203			SyncMode.PROXY: '(REMOTE ➔ LOCAL ➔ REMOTE)',
204			SyncMode.DUMP_LOCAL: '(LOCAL, ONLY EXPORT)',
205			SyncMode.DUMP_REMOTE: '(REMOTE, ONLY EXPORT)',
206			SyncMode.IMPORT_LOCAL: '(REMOTE, ONLY IMPORT)',
207			SyncMode.IMPORT_REMOTE: '(LOCAL, ONLY IMPORT)',
208			SyncMode.SYNC_LOCAL: '(LOCAL ➔ LOCAL)',
209			SyncMode.SYNC_REMOTE: '(REMOTE ➔ REMOTE)'
210			}
211
212			for _mode, _desc in _modes.items():
213			if getattr(SyncMode, 'is_' + _mode.lower())():
214			sync_mode = _mode
215			_description = _desc
216
217			if is_import():
218			output.message(
219			output.Subject.INFO,
220			f'Import file {output.CliFormat.BLACK}{system.config["import"]}{output.CliFormat.ENDC}',
221			True
222			)
223
224			system.config['is_same_client'] = SyncMode.is_same_host()
225
226			output.message(
227			output.Subject.INFO,
228			f'Sync mode: {sync_mode} {output.CliFormat.BLACK}{_description}{output.CliFormat.ENDC}',
229			True
230			)
231
232			check_for_protection()
233
234
235			def is_remote(client):
236			"""
237			Check if given client is remote client
238			:param client: String
239			:return: Boolean
240			"""
241			if client == Client.ORIGIN:
			0 ignored issues – show unused-code introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Unnecessary "elif" after "return" Loading history...
242			return is_origin_remote()
243			elif client == Client.TARGET:
244			return is_target_remote()
245			elif client == Client.LOCAL:
246			return False
247			else:
248			return False
249
250
251			def is_target_remote():
252			"""
253			Check if target is remote client
254			:return: Boolean
255			"""
256			return sync_mode in (SyncMode.SENDER, SyncMode.PROXY, SyncMode.DUMP_REMOTE,
257			SyncMode.IMPORT_REMOTE, SyncMode.SYNC_REMOTE)
258
259
260			def is_origin_remote():
261			"""
262			Check if origin is remote client
263			:return: Boolean
264			"""
265			return sync_mode in (SyncMode.RECEIVER, SyncMode.PROXY, SyncMode.DUMP_REMOTE,
266			SyncMode.IMPORT_REMOTE, SyncMode.SYNC_REMOTE)
267
268
269			def is_import():
270			"""
271			Check if sync mode is import
272			:return: Boolean
273			"""
274			return sync_mode in (SyncMode.IMPORT_LOCAL, SyncMode.IMPORT_REMOTE)
275
276
277			def is_dump():
278			"""
279			Check if sync mode is import
280			:return: Boolean
281			"""
282			return sync_mode in (SyncMode.DUMP_LOCAL, SyncMode.DUMP_REMOTE)
283
284
285			def run_command(command, client, force_output=False, allow_fail=False, skip_dry_run=False):
			0 ignored issues – show Unused Code introduced 2020-11-21 18:36 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...
286			"""
287			Run command depending on the given client
288			:param command: String
289			:param client: String
290			:param force_output: Boolean
291			:param allow_fail: Boolean
292			:param skip_dry_run: Boolean
293			:return:
294			"""
295			if system.config['verbose']:
296			output.message(
297			output.host_to_subject(client),
298			output.CliFormat.BLACK + command + output.CliFormat.ENDC,
299			debug=True
300			)
301
302			if system.config['dry_run'] and skip_dry_run:
303			return
304
305			if is_remote(client):
306			if force_output:
			0 ignored issues – show unused-code introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report Unnecessary "else" after "return" Loading history...
307			return ''.join(remote_system.run_ssh_command_by_client(client, command).readlines()).strip()
			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 (104/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
308			else:
309			return remote_system.run_ssh_command_by_client(client, command)
310			else:
311			res = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
312			# Wait for the process end and print error in case of failure
313			out, err = res.communicate()
314
315			if res.wait() != 0 and err.decode() != '' and not allow_fail:
316			helper.run_script(script='error')
317			sys.exit(output.message(output.Subject.ERROR, err.decode(), False))
318
319			if force_output:
320			return out.decode().strip()
321
322
323			def check_for_protection():
324			"""
325			Check if the target system is protected
326			:return: Boolean
327			"""
328			if sync_mode in (SyncMode.RECEIVER, SyncMode.SENDER, SyncMode.PROXY, SyncMode.SYNC_LOCAL,
329			SyncMode.SYNC_REMOTE, SyncMode.IMPORT_LOCAL, SyncMode.IMPORT_REMOTE) and \
330			'protect' in system.config[Client.TARGET]:
331			_host = helper.get_ssh_host_name(Client.TARGET)
332			sys.exit(output.message(output.Subject.ERROR,
333			f'The host {_host} is protected against the import of a database dump. Please '
			0 ignored issues – show Coding Style introduced 2022-04-01 16:16 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (111/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
334			'check synchronisation target or adjust the host configuration.', False))
			0 ignored issues – show Coding Style introduced 2022-04-01 16:16 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (105/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
335
			0 ignored issues – show coding-style introduced 2022-04-01 16:16 UTC by Report Bug Copy Issue Report Trailing newlines Loading history...
336

jackd248 / db-sync-tool

db_sync_tool.utility.mode.run_command() C last analyzed 2025-07-23 08:59 UTC

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like

db_sync_tool.utility.mode.run_command() C
last analyzed 2025-07-23 08:59 UTC