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

db_sync_tool.remote.transfer.get_sftp_client() A
last analyzed 2025-07-23 08:59 UTC

↳ Parent: db_sync_tool.remote.transfer

Complexity

Conditions

Size

Total Lines	9
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	9
rs	10
c	0
b	0
f	0
cc	1
nop	1

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

"""
Transfer script
"""

import sys
from db_sync_tool.utility import mode, system, helper, output
from db_sync_tool.database import utility as database_utility
from db_sync_tool.remote import utility, client, rsync


def transfer_origin_database_dump():
    """
    Transfer the origin database dump files
    :return:
    """
    if not mode.is_import():
        if mode.get_sync_mode() == mode.SyncMode.RECEIVER:
            get_origin_database_dump(helper.get_dump_dir(mode.Client.TARGET))
            system.check_target_configuration()
        elif mode.get_sync_mode() == mode.SyncMode.SENDER:
            system.check_target_configuration()
            put_origin_database_dump(helper.get_dump_dir(mode.Client.ORIGIN))
            utility.remove_origin_database_dump()
        elif mode.get_sync_mode() == mode.SyncMode.PROXY:
            helper.create_local_temporary_data_dir()
            get_origin_database_dump(system.default_local_sync_path)
            system.check_target_configuration()
            put_origin_database_dump(system.default_local_sync_path)
        elif mode.get_sync_mode() == mode.SyncMode.SYNC_REMOTE or mode.get_sync_mode() == mode.SyncMode.SYNC_LOCAL:

            system.check_target_configuration()
        elif system.config['is_same_client']:
            utility.remove_origin_database_dump(True)
    else:
        system.check_target_configuration()


def get_origin_database_dump(target_path):

    """
    Downloading the origin database dump files
    :param target_path: String
    :return:
    """
    output.message(
        output.Subject.ORIGIN,
        'Downloading database dump',
        True
    )
    if mode.get_sync_mode() != mode.SyncMode.PROXY:
        helper.check_and_create_dump_dir(mode.Client.TARGET, target_path)

    if not system.config['dry_run']:
        _remotepath = helper.get_dump_dir(mode.Client.ORIGIN) + database_utility.database_dump_file_name + '.tar.gz'

        _localpath = target_path

        if system.config['use_rsync']:
            rsync.run_rsync_command(
                remote_client=mode.Client.ORIGIN,
                origin_path=_remotepath,
                target_path=_localpath,
                origin_ssh=system.config[mode.Client.ORIGIN]['user'] + '@' + system.config[mode.Client.ORIGIN]['host']

            )
        else:
            #
            # Download speed problems
            # https://github.com/paramiko/paramiko/issues/60
            #
            sftp = get_sftp_client(client.ssh_client_origin)
            sftp.get(helper.get_dump_dir(mode.Client.ORIGIN) + database_utility.database_dump_file_name + '.tar.gz',

                     target_path + database_utility.database_dump_file_name + '.tar.gz', download_status)

            sftp.close()
            if not system.config['mute']:
                print('')

    utility.remove_origin_database_dump()


def download_status(sent, size):
    """
    Printing the download status information
    :param sent: Float
    :param size: Float
    :return:
    """
    if not system.config['mute']:
        sent_mb = round(float(sent) / 1024 / 1024, 1)
        size = round(float(size) / 1024 / 1024, 1)
        sys.stdout.write(
            output.Subject.ORIGIN + output.CliFormat.BLACK + '[REMOTE]' + output.CliFormat.ENDC + " Status: {0} MB of {1} MB downloaded".

            format(sent_mb, size, ))
        sys.stdout.write('\r')


def put_origin_database_dump(origin_path):

    """
    Uploading the origin database dump file
    :param origin_path: String
    :return:
    """
    if mode.get_sync_mode() == mode.SyncMode.PROXY:
        _subject = output.Subject.LOCAL
    else:
        _subject = output.Subject.ORIGIN

    output.message(
        _subject,
        'Uploading database dump',
        True
    )
    helper.check_and_create_dump_dir(mode.Client.TARGET, helper.get_dump_dir(mode.Client.TARGET))

    if not system.config['dry_run']:
        _localpath = origin_path + database_utility.database_dump_file_name + '.tar.gz'
        _remotepath = helper.get_dump_dir(mode.Client.TARGET) + '/'

        if system.config['use_rsync']:
            rsync.run_rsync_command(
                remote_client=mode.Client.TARGET,
                origin_path=_localpath,
                target_path=_remotepath,
                target_ssh=system.config[mode.Client.TARGET]['user'] + '@' + system.config[mode.Client.TARGET]['host']

            )
        else:
            #
            # Download speed problems
            # https://github.com/paramiko/paramiko/issues/60
            #
            sftp = get_sftp_client(client.ssh_client_target)
            sftp.put(origin_path + database_utility.database_dump_file_name + '.tar.gz',
                     helper.get_dump_dir(mode.Client.TARGET) + database_utility.database_dump_file_name + '.tar.gz',

                     upload_status)
            sftp.close()
            if not system.config['mute']:
                print('')



def upload_status(sent, size):
    """
    Printing the upload status information
    :param sent: Float
    :param size: Float
    :return:
    """
    if not system.config['mute']:
        sent_mb = round(float(sent) / 1024 / 1024, 1)
        size = round(float(size) / 1024 / 1024, 1)

        if (mode.get_sync_mode() == mode.SyncMode.PROXY):

            _subject = output.Subject.LOCAL
        else:
            _subject = output.Subject.ORIGIN + output.CliFormat.BLACK + '[LOCAL]' + output.CliFormat.ENDC


        sys.stdout.write(
            _subject + " Status: {0} MB of {1} MB uploaded".
            format(sent_mb, size, ))
        sys.stdout.write('\r')


def get_sftp_client(ssh_client):
    """

    :param ssh_client:
    :return:
    """
    sftp = ssh_client.open_sftp()
    sftp.get_channel().settimeout(client.default_timeout)
    return sftp




1		#!/usr/bin/env python3
2		# -- coding: future_fstrings --
3
4		"""
5		Transfer script
6		"""
7
8		import sys
9		from db_sync_tool.utility import mode, system, helper, output
10		from db_sync_tool.database import utility as database_utility
11		from db_sync_tool.remote import utility, client, rsync
12
13
14		def transfer_origin_database_dump():
15		"""
16		Transfer the origin database dump files
17		:return:
18		"""
19		if not mode.is_import():
20		if mode.get_sync_mode() == mode.SyncMode.RECEIVER:
21		get_origin_database_dump(helper.get_dump_dir(mode.Client.TARGET))
22		system.check_target_configuration()
23		elif mode.get_sync_mode() == mode.SyncMode.SENDER:
24		system.check_target_configuration()
25		put_origin_database_dump(helper.get_dump_dir(mode.Client.ORIGIN))
26		utility.remove_origin_database_dump()
27		elif mode.get_sync_mode() == mode.SyncMode.PROXY:
28		helper.create_local_temporary_data_dir()
29		get_origin_database_dump(system.default_local_sync_path)
30		system.check_target_configuration()
31		put_origin_database_dump(system.default_local_sync_path)
32		elif mode.get_sync_mode() == mode.SyncMode.SYNC_REMOTE or mode.get_sync_mode() == mode.SyncMode.SYNC_LOCAL:
		0 ignored issues – show Coding Style introduced 2021-03-07 13:09 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (115/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
33		system.check_target_configuration()
34		elif system.config['is_same_client']:
35		utility.remove_origin_database_dump(True)
36		else:
37		system.check_target_configuration()
38
39
40	View Code Duplication	def get_origin_database_dump(target_path):
		0 ignored issues – show Duplication introduced 2022-02-13 19:37 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
41		"""
42		Downloading the origin database dump files
43		:param target_path: String
44		:return:
45		"""
46		output.message(
47		output.Subject.ORIGIN,
48		'Downloading database dump',
49		True
50		)
51		if mode.get_sync_mode() != mode.SyncMode.PROXY:
52		helper.check_and_create_dump_dir(mode.Client.TARGET, target_path)
53
54		if not system.config['dry_run']:
55		_remotepath = helper.get_dump_dir(mode.Client.ORIGIN) + database_utility.database_dump_file_name + '.tar.gz'
		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 (116/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
56		_localpath = target_path
57
58		if system.config['use_rsync']:
59		rsync.run_rsync_command(
60		remote_client=mode.Client.ORIGIN,
61		origin_path=_remotepath,
62		target_path=_localpath,
63		origin_ssh=system.config[mode.Client.ORIGIN]['user'] + '@' + system.config[mode.Client.ORIGIN]['host']
		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 (118/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
64		)
65		else:
66		#
67		# Download speed problems
68		# https://github.com/paramiko/paramiko/issues/60
69		#
70		sftp = get_sftp_client(client.ssh_client_origin)
71		sftp.get(helper.get_dump_dir(mode.Client.ORIGIN) + database_utility.database_dump_file_name + '.tar.gz',
		0 ignored issues – show Coding Style introduced 2020-11-27 11:55 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (116/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
72		target_path + database_utility.database_dump_file_name + '.tar.gz', download_status)
		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 (105/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
73		sftp.close()
74		if not system.config['mute']:
75		print('')
76
77		utility.remove_origin_database_dump()
78
79
80		def download_status(sent, size):
81		"""
82		Printing the download status information
83		:param sent: Float
84		:param size: Float
85		:return:
86		"""
87		if not system.config['mute']:
88		sent_mb = round(float(sent) / 1024 / 1024, 1)
89		size = round(float(size) / 1024 / 1024, 1)
90		sys.stdout.write(
91		output.Subject.ORIGIN + output.CliFormat.BLACK + '[REMOTE]' + output.CliFormat.ENDC + " Status: {0} MB of {1} MB downloaded".
		0 ignored issues – show Coding Style introduced 2021-03-06 18:29 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (137/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
92		format(sent_mb, size, ))
93		sys.stdout.write('\r')
94
95
96	View Code Duplication	def put_origin_database_dump(origin_path):
		0 ignored issues – show Duplication introduced 2022-02-13 19:37 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
97		"""
98		Uploading the origin database dump file
99		:param origin_path: String
100		:return:
101		"""
102		if mode.get_sync_mode() == mode.SyncMode.PROXY:
103		_subject = output.Subject.LOCAL
104		else:
105		_subject = output.Subject.ORIGIN
106
107		output.message(
108		_subject,
109		'Uploading database dump',
110		True
111		)
112		helper.check_and_create_dump_dir(mode.Client.TARGET, helper.get_dump_dir(mode.Client.TARGET))
113
114		if not system.config['dry_run']:
115		_localpath = origin_path + database_utility.database_dump_file_name + '.tar.gz'
116		_remotepath = helper.get_dump_dir(mode.Client.TARGET) + '/'
117
118		if system.config['use_rsync']:
119		rsync.run_rsync_command(
120		remote_client=mode.Client.TARGET,
121		origin_path=_localpath,
122		target_path=_remotepath,
123		target_ssh=system.config[mode.Client.TARGET]['user'] + '@' + system.config[mode.Client.TARGET]['host']
		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 (118/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
124		)
125		else:
126		#
127		# Download speed problems
128		# https://github.com/paramiko/paramiko/issues/60
129		#
130		sftp = get_sftp_client(client.ssh_client_target)
131		sftp.put(origin_path + database_utility.database_dump_file_name + '.tar.gz',
132		helper.get_dump_dir(mode.Client.TARGET) + database_utility.database_dump_file_name + '.tar.gz',
		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 (116/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
133		upload_status)
134		sftp.close()
135		if not system.config['mute']:
136		print('')
137
138
139
140		def upload_status(sent, size):
141		"""
142		Printing the upload status information
143		:param sent: Float
144		:param size: Float
145		:return:
146		"""
147		if not system.config['mute']:
148		sent_mb = round(float(sent) / 1024 / 1024, 1)
149		size = round(float(size) / 1024 / 1024, 1)
150
151		if (mode.get_sync_mode() == mode.SyncMode.PROXY):
		0 ignored issues – show Unused Code Coding Style introduced 2020-11-21 18:36 UTC by Report Bug Copy Issue Report There is an unnecessary parenthesis after `if`. Loading history...
152		_subject = output.Subject.LOCAL
153		else:
154		_subject = output.Subject.ORIGIN + output.CliFormat.BLACK + '[LOCAL]' + output.CliFormat.ENDC
		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 (105/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
155
156		sys.stdout.write(
157		_subject + " Status: {0} MB of {1} MB uploaded".
158		format(sent_mb, size, ))
159		sys.stdout.write('\r')
160
161
162		def get_sftp_client(ssh_client):
163		"""
164
165		:param ssh_client:
166		:return:
167		"""
168		sftp = ssh_client.open_sftp()
169		sftp.get_channel().settimeout(client.default_timeout)
170		return sftp
171
		0 ignored issues – show coding-style introduced 2022-02-13 19:37 UTC by Report Bug Copy Issue Report Trailing newlines Loading history...
172

jackd248 / db-sync-tool

db_sync_tool.remote.transfer.get_sftp_client() A last analyzed 2025-07-23 08:59 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

db_sync_tool.remote.transfer.get_sftp_client() A
last analyzed 2025-07-23 08:59 UTC