db_sync_tool.remote.rsync.unit_converter() - Code Metrics - Inspection of "[Release] Version 2.6.0" - jackd248/db-sync-tool - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 190e21...7fdd2b )

by Konrad

created 2022-02-13 19:28 UTC

db_sync_tool.remote.rsync.unit_converter() A

↳ Parent: db_sync_tool.remote.rsync

Complexity

Conditions

Size

Total Lines	15
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	8
dl	0
loc	15
rs	10
c	0
b	0
f	0
cc	3
nop	1

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

"""
rsync script
"""

import re
from db_sync_tool.utility import mode, system, output

# Default options for rsync command
# https://wiki.ubuntuusers.de/rsync/
default_options = [
    '--delete',
    '-a',
    '-z',
    '--stats',
    '--human-readable',
    '--iconv=UTF-8',
    '--chmod=D2770,F660'
]

def get_password_environment(client):
    """
    Optionally create a password environment variable for sshpass password authentication
    https://www.redhat.com/sysadmin/ssh-automation-sshpass
    :param client: String
    :return:
    """
    if not client:
        return ''

    if system.config['use_sshpass'] and not 'ssh_key' in system.config[client] and 'password' in system.config[client]:

        return f'SSHPASS=\'{system.config[client]["password"]}\' '
    return ''


def get_authorization(client):
    """
    Define authorization arguments for rsync command
    :param client: String
    :return: String
    """
    _ssh_key = None
    if not client:
        return ''

    if 'ssh_key' in system.config[client]:
        _ssh_key = system.config[mode.Client.ORIGIN]['ssh_key']

    _ssh_port = system.config[client]['port'] if 'port' in system.config[client] else 22

    if _ssh_key is None:
        if system.config['use_sshpass'] and get_password_environment(client):

            # In combination with SSHPASS environment variable
            # https://www.redhat.com/sysadmin/ssh-automation-sshpass
            return f'--rsh="sshpass -e ssh -p{_ssh_port} -o StrictHostKeyChecking=no -l {system.config[client]["user"]}"'

        else:
            return f'-e "ssh -p{_ssh_port} -o StrictHostKeyChecking=no"'
    else:
        # Provide ssh key file path for ssh authentication
        return f'-e "ssh -i {_ssh_key} -p{_ssh_port}"'


def get_host(client):
    """
    Return user@host if client is not local
    :param client: String
    :return: String
    """
    if mode.is_remote(client):
        return f'{system.config[client]["user"]}@{system.config[client]["host"]}:'
    return ''


def get_options():
    """
    Prepare rsync options with stored default options and provided addtional options
    :return: String
    """
    _options = f'{" ".join(default_options)}'
    if not system.config['use_rsync_options'] is None:
        _options += f'{system.config["use_rsync_options"]}'
    return _options


def read_stats(stats):
    """
    Read rsync stats and print a summary
    :param stats: String
    :return:
    """
    if system.config['verbose']:
        print(f'{output.Subject.DEBUG}{output.CliFormat.BLACK}{stats}{output.CliFormat.ENDC}')

    _file_size = parse_string(stats, r'Total transferred file size:\s*([\d.]+[MKG]?)')

    if _file_size:
        output.message(
            output.Subject.INFO,
            f'Status: {unit_converter(_file_size[0])} transferred'
        )


def parse_string(string, regex):
    """
    Parse string by given regex
    :param string: String
    :param regex: String
    :return:
    """
    _file_size_pattern = regex
    _regex_matcher = re.compile(_file_size_pattern)
    return _regex_matcher.findall(string)


def unit_converter(size_in_bytes):
    """

    :param size_in_bytes:
    :return:
    """
    units = ['Bytes', 'kB', 'MB', 'GB']

    _convertedSize = float(size_in_bytes)

    for unit in units:
        if _convertedSize < 1024:
            return str(_convertedSize) + ' ' + unit
        _convertedSize = _convertedSize/1024


    return _convertedSize


def run_rsync_command(remote_client, origin_path, target_path, origin_ssh = '', target_ssh = ''):

    """

    :param localpath:
    :param remotepath:
    :return:
    """
    if origin_ssh != '':
        origin_ssh += ':'
    if target_ssh != '':
        target_ssh += ':'

    _output = mode.run_command(
        f'{get_password_environment(remote_client)}rsync {get_options()} '
        f'{get_authorization(remote_client)} '
        f'{origin_ssh}{origin_path} {target_ssh}{target_path}',
        mode.Client.LOCAL,
        True
    )
    read_stats(_output)




1			#!/usr/bin/env python3
2			# -- coding: future_fstrings --
3
4			"""
5			rsync script
6			"""
7
8			import re
9			from db_sync_tool.utility import mode, system, output
10
11			# Default options for rsync command
12			# https://wiki.ubuntuusers.de/rsync/
13			default_options = [
14			'--delete',
15			'-a',
16			'-z',
17			'--stats',
18			'--human-readable',
19			'--iconv=UTF-8',
20			'--chmod=D2770,F660'
21			]
22
23			def get_password_environment(client):
24			"""
25			Optionally create a password environment variable for sshpass password authentication
26			https://www.redhat.com/sysadmin/ssh-automation-sshpass
27			:param client: String
28			:return:
29			"""
30			if not client:
31			return ''
32
33			if system.config['use_sshpass'] and not 'ssh_key' in system.config[client] and 'password' in system.config[client]:
			0 ignored issues – show Coding Style introduced 2022-02-13 19:37 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (119/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
34			return f'SSHPASS=\'{system.config[client]["password"]}\' '
35			return ''
36
37
38			def get_authorization(client):
39			"""
40			Define authorization arguments for rsync command
41			:param client: String
42			:return: String
43			"""
44			_ssh_key = None
45			if not client:
46			return ''
47
48			if 'ssh_key' in system.config[client]:
49			_ssh_key = system.config[mode.Client.ORIGIN]['ssh_key']
50
51			_ssh_port = system.config[client]['port'] if 'port' in system.config[client] else 22
52
53			if _ssh_key is None:
54			if system.config['use_sshpass'] and get_password_environment(client):
			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...
55			# In combination with SSHPASS environment variable
56			# https://www.redhat.com/sysadmin/ssh-automation-sshpass
57			return f'--rsh="sshpass -e ssh -p{_ssh_port} -o StrictHostKeyChecking=no -l {system.config[client]["user"]}"'
			0 ignored issues – show Coding Style introduced 2022-02-13 19:37 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (121/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
58			else:
59			return f'-e "ssh -p{_ssh_port} -o StrictHostKeyChecking=no"'
60			else:
61			# Provide ssh key file path for ssh authentication
62			return f'-e "ssh -i {_ssh_key} -p{_ssh_port}"'
63
64
65			def get_host(client):
66			"""
67			Return user@host if client is not local
68			:param client: String
69			:return: String
70			"""
71			if mode.is_remote(client):
72			return f'{system.config[client]["user"]}@{system.config[client]["host"]}:'
73			return ''
74
75
76			def get_options():
77			"""
78			Prepare rsync options with stored default options and provided addtional options
79			:return: String
80			"""
81			_options = f'{" ".join(default_options)}'
82			if not system.config['use_rsync_options'] is None:
83			_options += f'{system.config["use_rsync_options"]}'
84			return _options
85
86
87			def read_stats(stats):
88			"""
89			Read rsync stats and print a summary
90			:param stats: String
91			:return:
92			"""
93			if system.config['verbose']:
94			print(f'{output.Subject.DEBUG}{output.CliFormat.BLACK}{stats}{output.CliFormat.ENDC}')
95
96			_file_size = parse_string(stats, r'Total transferred file size:\s*([\d.]+[MKG]?)')
97
98			if _file_size:
99			output.message(
100			output.Subject.INFO,
101			f'Status: {unit_converter(_file_size[0])} transferred'
102			)
103
104
105			def parse_string(string, regex):
106			"""
107			Parse string by given regex
108			:param string: String
109			:param regex: String
110			:return:
111			"""
112			_file_size_pattern = regex
113			_regex_matcher = re.compile(_file_size_pattern)
114			return _regex_matcher.findall(string)
115
116
117			def unit_converter(size_in_bytes):
118			"""
119
120			:param size_in_bytes:
121			:return:
122			"""
123			units = ['Bytes', 'kB', 'MB', 'GB']
124
125			_convertedSize = float(size_in_bytes)
			0 ignored issues – show Coding Style Naming introduced 2022-02-13 19:37 UTC by Report Bug Copy Issue Report Variable name "_convertedSize" 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...
126			for unit in units:
127			if _convertedSize < 1024:
128			return str(_convertedSize) + ' ' + unit
129			_convertedSize = _convertedSize/1024
			0 ignored issues – show Coding Style Naming introduced 2022-02-13 19:37 UTC by Report Bug Copy Issue Report Variable name "_convertedSize" 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...
130
131			return _convertedSize
132
133
134			def run_rsync_command(remote_client, origin_path, target_path, origin_ssh = '', target_ssh = ''):
			0 ignored issues – show Coding Style introduced 2022-02-13 19:37 UTC by Report Bug Copy Issue Report No space allowed around keyword argument assignment Loading history...
135			"""
136
137			:param localpath:
138			:param remotepath:
139			:return:
140			"""
141			if origin_ssh != '':
142			origin_ssh += ':'
143			if target_ssh != '':
144			target_ssh += ':'
145
146			_output = mode.run_command(
147			f'{get_password_environment(remote_client)}rsync {get_options()} '
148			f'{get_authorization(remote_client)} '
149			f'{origin_ssh}{origin_path} {target_ssh}{target_path}',
150			mode.Client.LOCAL,
151			True
152			)
153			read_stats(_output)
154
			0 ignored issues – show coding-style introduced 2022-02-13 19:37 UTC by Report Bug Copy Issue Report Trailing newlines Loading history...
155

jackd248 / db-sync-tool

Push — master ( 190e21...7fdd2b )

db_sync_tool.remote.rsync.unit_converter() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like