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

Passed

Push — master ( b52b24...0918fa )

by Konrad

created 2024-03-20 10:26 UTC

remove_target_database_dump() B

↳ Parent: db_sync_tool.remote.utility

Complexity

Conditions

Size

Total Lines	52
Code Lines	32

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	32
dl	0
loc	52
rs	7.2453
c	0
b	0
f	0
cc	8
nop	0

How to fix Long Method

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

"""
Utility script
"""

import os
import paramiko
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 client as remote_client



def remove_origin_database_dump(keep_compressed_file=False):
    """
    Removing the origin database dump files
    :param keep_compressed_file: Boolean
    :return:
    """
    output.message(
        output.Subject.ORIGIN,
        'Cleaning up',
        True
    )

    if system.config['dry_run']:
        return

    _file_path = helper.get_dump_dir(mode.Client.ORIGIN) + database_utility.database_dump_file_name
    if mode.is_origin_remote():
        mode.run_command(
            helper.get_command(mode.Client.ORIGIN, 'rm') + ' ' + _file_path,
            mode.Client.ORIGIN
        )
        if not keep_compressed_file:
            mode.run_command(
                helper.get_command(mode.Client.ORIGIN, 'rm') + ' ' + _file_path + '.tar.gz',
                mode.Client.ORIGIN
            )
    else:
        os.remove(_file_path)
        if not keep_compressed_file:
            os.remove(f'{_file_path}.tar.gz')

    if keep_compressed_file:
        if 'keep_dumps' in system.config[mode.Client.ORIGIN]:
            helper.clean_up_dump_dir(mode.Client.ORIGIN,
                                     helper.get_dump_dir(mode.Client.ORIGIN) + '*',
                                     system.config[mode.Client.ORIGIN]['keep_dumps'])

        output.message(
            output.Subject.INFO,
            f'Database dump file is saved to: {_file_path}.tar.gz',
            True,
            True
        )


def remove_target_database_dump():
    """
    Removing the target database dump files
    :return:
    """
    _file_path = helper.get_dump_dir(mode.Client.TARGET) + database_utility.database_dump_file_name

    #
    # Move dump to specified directory
    #
    if system.config['keep_dump']:
        helper.create_local_temporary_data_dir()
        _keep_dump_path = system.default_local_sync_path + database_utility.database_dump_file_name
        mode.run_command(
            helper.get_command('target',
                               'cp') + ' ' + _file_path + ' ' + _keep_dump_path,
            mode.Client.TARGET
        )
        output.message(
            output.Subject.INFO,
            f'Database dump file is saved to: {_keep_dump_path}',
            True,
            True
        )

    #
    # Clean up
    #
    if not mode.is_dump() and not mode.is_import():
        output.message(
            output.Subject.TARGET,
            'Cleaning up',
            True
        )

        if system.config['dry_run']:
            return

        if mode.is_target_remote():
            mode.run_command(
                helper.get_command(mode.Client.TARGET, 'rm') + ' ' + _file_path,
                mode.Client.TARGET
            )
            mode.run_command(
                helper.get_command(mode.Client.TARGET, 'rm') + ' ' + _file_path + '.tar.gz',
                mode.Client.TARGET
            )
        else:
            if os.path.isfile(_file_path):
                os.remove(_file_path)
            if os.path.isfile(f'{_file_path}.tar.gz'):
                os.remove(f'{_file_path}.tar.gz')


def check_keys_from_ssh_agent():
    """
    Check if private keys are available from an SSH agent.
    :return:
    """
    agent = paramiko.Agent()
    agent_keys = agent.get_keys()
    if len(agent_keys) == 0:
        return False
    return True


1			#!/usr/bin/env python3
2			# -- coding: future_fstrings --
3
4			"""
5			Utility script
6			"""
7
8			import os
9			import paramiko
10			from db_sync_tool.utility import mode, system, helper, output
11			from db_sync_tool.database import utility as database_utility
12			from db_sync_tool.remote import client as remote_client
			0 ignored issues – show Unused Code introduced 2024-03-20 10:28 UTC by Report Bug Copy Issue Report Unused client imported from db_sync_tool.remote as remote_client Loading history...
13
14
15			def remove_origin_database_dump(keep_compressed_file=False):
16			"""
17			Removing the origin database dump files
18			:param keep_compressed_file: Boolean
19			:return:
20			"""
21			output.message(
22			output.Subject.ORIGIN,
23			'Cleaning up',
24			True
25			)
26
27			if system.config['dry_run']:
28			return
29
30			_file_path = helper.get_dump_dir(mode.Client.ORIGIN) + database_utility.database_dump_file_name
31			if mode.is_origin_remote():
32			mode.run_command(
33			helper.get_command(mode.Client.ORIGIN, 'rm') + ' ' + _file_path,
34			mode.Client.ORIGIN
35			)
36			if not keep_compressed_file:
37			mode.run_command(
38			helper.get_command(mode.Client.ORIGIN, 'rm') + ' ' + _file_path + '.tar.gz',
39			mode.Client.ORIGIN
40			)
41			else:
42			os.remove(_file_path)
43			if not keep_compressed_file:
44			os.remove(f'{_file_path}.tar.gz')
45
46			if keep_compressed_file:
47			if 'keep_dumps' in system.config[mode.Client.ORIGIN]:
48			helper.clean_up_dump_dir(mode.Client.ORIGIN,
49			helper.get_dump_dir(mode.Client.ORIGIN) + '*',
50			system.config[mode.Client.ORIGIN]['keep_dumps'])
51
52			output.message(
53			output.Subject.INFO,
54			f'Database dump file is saved to: {_file_path}.tar.gz',
55			True,
56			True
57			)
58
59
60			def remove_target_database_dump():
61			"""
62			Removing the target database dump files
63			:return:
64			"""
65			_file_path = helper.get_dump_dir(mode.Client.TARGET) + database_utility.database_dump_file_name
66
67			#
68			# Move dump to specified directory
69			#
70			if system.config['keep_dump']:
71			helper.create_local_temporary_data_dir()
72			_keep_dump_path = system.default_local_sync_path + database_utility.database_dump_file_name
73			mode.run_command(
74			helper.get_command('target',
75			'cp') + ' ' + _file_path + ' ' + _keep_dump_path,
76			mode.Client.TARGET
77			)
78			output.message(
79			output.Subject.INFO,
80			f'Database dump file is saved to: {_keep_dump_path}',
81			True,
82			True
83			)
84
85			#
86			# Clean up
87			#
88			if not mode.is_dump() and not mode.is_import():
89			output.message(
90			output.Subject.TARGET,
91			'Cleaning up',
92			True
93			)
94
95			if system.config['dry_run']:
96			return
97
98			if mode.is_target_remote():
99			mode.run_command(
100			helper.get_command(mode.Client.TARGET, 'rm') + ' ' + _file_path,
101			mode.Client.TARGET
102			)
103			mode.run_command(
104			helper.get_command(mode.Client.TARGET, 'rm') + ' ' + _file_path + '.tar.gz',
105			mode.Client.TARGET
106			)
107			else:
108			if os.path.isfile(_file_path):
109			os.remove(_file_path)
110			if os.path.isfile(f'{_file_path}.tar.gz'):
111			os.remove(f'{_file_path}.tar.gz')
112
113
114			def check_keys_from_ssh_agent():
115			"""
116			Check if private keys are available from an SSH agent.
117			:return:
118			"""
119			agent = paramiko.Agent()
120			agent_keys = agent.get_keys()
121			if len(agent_keys) == 0:
122			return False
123			return True
124

jackd248 / db-sync-tool

Push — master ( b52b24...0918fa )

remove_target_database_dump() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like