db_sync_tool.remote.system.run_ssh_command_by_client() - Code Metrics - jackd248/db-sync-tool - Measure and Improve Code Quality continuously with Scrutinizer

run_ssh_command_by_client() A
last analyzed 2025-04-09 15:26 UTC

↳ Parent: db_sync_tool.remote.system

Complexity

Conditions

Size

Total Lines	11
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	11
rs	10
c	0
b	0
f	0
cc	2
nop	2

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

"""
System script
"""

import sys
from db_sync_tool.utility import mode, output, helper
from db_sync_tool.remote import client as remote_client


def run_ssh_command_by_client(client, command):
    """
    Running origin ssh command
    :param client: String
    :param command: String
    :return:
    """
    if client == mode.Client.ORIGIN:

        return run_ssh_command(command, remote_client.ssh_client_origin, client)
    else:
        return run_ssh_command(command, remote_client.ssh_client_target, client)


def run_ssh_command(command, ssh_client=remote_client.ssh_client_origin, client=None):
    """
    Running ssh command
    :param command: String
    :param ssh_client:
    :param client: String
    :return:
    """
    stdin, stdout, stderr = ssh_client.exec_command(command)

    exit_status = stdout.channel.recv_exit_status()

    err = stderr.read().decode()

    if err and exit_status != 0:
        helper.run_script(client=client, script='error')
        sys.exit(output.message(output.Subject.ERROR, err, False))
    elif err:
        output.message(output.Subject.WARNING, err, True)

    return stdout


1			#!/usr/bin/env python3
2			# -- coding: future_fstrings --
3
4			"""
5			System script
6			"""
7
8			import sys
9			from db_sync_tool.utility import mode, output, helper
10			from db_sync_tool.remote import client as remote_client
11
12
13			def run_ssh_command_by_client(client, command):
14			"""
15			Running origin ssh command
16			:param client: String
17			:param command: String
18			:return:
19			"""
20			if client == mode.Client.ORIGIN:
			0 ignored issues – show unused-code introduced 2021-04-13 18:13 UTC by Report Bug Copy Issue Report Unnecessary "else" after "return" Loading history...
21			return run_ssh_command(command, remote_client.ssh_client_origin, client)
22			else:
23			return run_ssh_command(command, remote_client.ssh_client_target, client)
24
25
26			def run_ssh_command(command, ssh_client=remote_client.ssh_client_origin, client=None):
27			"""
28			Running ssh command
29			:param command: String
30			:param ssh_client:
31			:param client: String
32			:return:
33			"""
34			stdin, stdout, stderr = ssh_client.exec_command(command)
			0 ignored issues – show Unused Code introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report The variable `stdin` seems to be unused. Loading history...
35			exit_status = stdout.channel.recv_exit_status()
36
37			err = stderr.read().decode()
38
39			if err and exit_status != 0:
40			helper.run_script(client=client, script='error')
41			sys.exit(output.message(output.Subject.ERROR, err, False))
42			elif err:
43			output.message(output.Subject.WARNING, err, True)
44
45			return stdout
46

jackd248 / db-sync-tool

run_ssh_command_by_client() A last analyzed 2025-04-09 15:26 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

run_ssh_command_by_client() A
last analyzed 2025-04-09 15:26 UTC