pystratum_mysql.backend.MySqlRoutineLoaderWorker.MySqlRoutineLoaderWorker.__init__() - Code Metrics - Inspection of "Release: 1.1.0" - DatabaseStratum/py-stratum-mysql - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 7c9493...8a6195 )

by P.R.

created 2022-01-30 12:45 UTC

MySqlRoutineLoaderWorker.init() A

↳ Parent: pystratum_mysql.backend.MySqlRoutineLoaderWorker

Complexity

Conditions

Size

Total Lines	21
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.6296

Importance

Changes

Metric	Value
eloc	6
dl	0
loc	21
ccs	1
cts	7
cp	0.1429
rs	10
c	0
b	0
f	0
cc	1
nop	3
crap	1.6296

from configparser import ConfigParser
from typing import Any, Dict, List, Optional

from pystratum_backend.StratumStyle import StratumStyle

from pystratum_common.backend.CommonRoutineLoaderWorker import CommonRoutineLoaderWorker
from pystratum_mysql.backend.MySqlWorker import MySqlWorker
from pystratum_mysql.helper.MySqlRoutineLoaderHelper import MySqlRoutineLoaderHelper


class MySqlRoutineLoaderWorker(MySqlWorker, CommonRoutineLoaderWorker):
    """
    Class for loading stored routines into a MySQL instance from (pseudo) SQL files.
    """
    MAX_LENGTH_CHAR = 255
    """
    Maximum length of a varchar.
    """

    MAX_LENGTH_VARCHAR = 4096
    """
    Maximum length of a varchar.
    """

    MAX_LENGTH_BINARY = 255
    """
    Maximum length of a varbinary.
    """

    MAX_LENGTH_VARBINARY = 4096
    """
    Maximum length of a varbinary.
    """

    # ------------------------------------------------------------------------------------------------------------------
    def __init__(self, io: StratumStyle, config: ConfigParser):
        """
        Object constructor.

        :param PyStratumStyle io: The output decorator.
        """
        MySqlWorker.__init__(self, io, config)
        CommonRoutineLoaderWorker.__init__(self, io, config)

        self._character_set_client: Optional[str] = None
        """
        The default character set under which the stored routine will be loaded and run.
        """

        self._collation_connection: Optional[str] = None
        """
        The default collate under which the stored routine will be loaded and run.
        """

        self._sql_mode: Optional[str] = None
        """
        
        """

    # ------------------------------------------------------------------------------------------------------------------
    def __save_column_types_exact(self, rows: List[Dict[str, Any]]) -> None:
        """
        Saves the exact column types as replace pairs.

        :param list[dict[str,*]] rows: The columns types.
        """
        for row in rows:
            key = row['table_name'] + '.' + row['column_name'] + '%type'

            value = row['column_type']
            if row['character_set_name']:
                value += ' character set ' + row['character_set_name']

            self._add_replace_pair(key, value, False)

    # ------------------------------------------------------------------------------------------------------------------
    def __save_column_types_max_length(self, rows: List[Dict[str, Any]]) -> None:
        """
        Saves the column types with maximum length as replace pairs.

        :param list[dict[str,*]] rows: The columns types.
        """
        for row in rows:
            key = row['table_name'] + '.' + row['column_name'] + '%max-type'

            if row['data_type'] == 'char':
                value = row['data_type'] + '(' + str(self.MAX_LENGTH_CHAR) + ')'
                value += ' character set ' + row['character_set_name']
                self._add_replace_pair(key, value, False)

            if row['data_type'] == 'varchar':
                value = row['data_type'] + '(' + str(self.MAX_LENGTH_VARCHAR) + ')'
                value += ' character set ' + row['character_set_name']
                self._add_replace_pair(key, value, False)

            elif row['data_type'] == 'binary':
                value = row['data_type'] + '(' + str(self.MAX_LENGTH_BINARY) + ')'
                self._add_replace_pair(key, value, False)

            elif row['data_type'] == 'varbinary':
                value = row['data_type'] + '(' + str(self.MAX_LENGTH_VARBINARY) + ')'
                self._add_replace_pair(key, value, False)

    # ------------------------------------------------------------------------------------------------------------------
    def _get_column_type(self) -> None:
        """
        Selects schema, table, column names and the column type from MySQL and saves them as replace pairs.
        """
        rows = self._dl.get_all_table_columns()
        self.__save_column_types_exact(rows)
        self.__save_column_types_max_length(rows)

        self._io.text('Selected {0} column types for substitution'.format(len(rows)))

    # ------------------------------------------------------------------------------------------------------------------
    def create_routine_loader_helper(self,
                                     routine_name: str,
                                     pystratum_old_metadata: Optional[Dict],
                                     rdbms_old_metadata: Optional[Dict]) -> MySqlRoutineLoaderHelper:
        """
        Creates a Routine Loader Helper object.

        :param str routine_name: The name of the routine.
        :param dict pystratum_old_metadata: The old metadata of the stored routine from PyStratum.
        :param dict rdbms_old_metadata:  The old metadata of the stored routine from MySQL.

        :rtype: MySqlRoutineLoaderHelper
        """
        return MySqlRoutineLoaderHelper(self._io,
                                        self._dl,
                                        self._source_file_names[routine_name],
                                        self._source_file_encoding,
                                        pystratum_old_metadata,
                                        self._replace_pairs,
                                        rdbms_old_metadata,
                                        self._sql_mode,
                                        self._character_set_client,
                                        self._collation_connection)

    # ------------------------------------------------------------------------------------------------------------------
    def _get_old_stored_routine_info(self) -> None:
        """
        Retrieves information about all stored routines in the current schema.
        """
        rows = self._dl.get_routines()
        self._rdbms_old_metadata = {}
        for row in rows:
            self._rdbms_old_metadata[row['routine_name']] = row

    # ------------------------------------------------------------------------------------------------------------------
    def _get_correct_sql_mode(self) -> None:
        """
        Gets the SQL mode in the order as preferred by MySQL.
        """
        self._sql_mode = self._dl.get_correct_sql_mode(self._sql_mode)

    # ------------------------------------------------------------------------------------------------------------------
    def _drop_obsolete_routines(self) -> None:
        """
        Drops obsolete stored routines (i.e. stored routines that exits in the current schema but for
        which we don't have a source file).
        """
        for routine_name, values in self._rdbms_old_metadata.items():
            if routine_name not in self._source_file_names:
                self._io.text("Dropping {0} <dbo>{1}</dbo>".format(values['routine_type'].lower(), routine_name))
                self._dl.drop_stored_routine(values['routine_type'], routine_name)

    # ------------------------------------------------------------------------------------------------------------------
    def _read_configuration_file(self) -> None:
        """
        Reads parameters from the configuration file.
        """
        CommonRoutineLoaderWorker._read_configuration_file(self)

        self._character_set_client = self._config.get('database', 'character_set_client', fallback='utf-8')
        self._collation_connection = self._config.get('database', 'collation_connection', fallback='utf8_general_ci')
        self._sql_mode = self._config.get('database', 'sql_mode')

# ----------------------------------------------------------------------------------------------------------------------


1	1	from configparser import ConfigParser
2	1	from typing import Any, Dict, List, Optional
3
4	1	from pystratum_backend.StratumStyle import StratumStyle
5
6	1	from pystratum_common.backend.CommonRoutineLoaderWorker import CommonRoutineLoaderWorker
7	1	from pystratum_mysql.backend.MySqlWorker import MySqlWorker
8	1	from pystratum_mysql.helper.MySqlRoutineLoaderHelper import MySqlRoutineLoaderHelper
9
10
11	1	class MySqlRoutineLoaderWorker(MySqlWorker, CommonRoutineLoaderWorker):
12		"""
13		Class for loading stored routines into a MySQL instance from (pseudo) SQL files.
14		"""
15	1	MAX_LENGTH_CHAR = 255
16		"""
17		Maximum length of a varchar.
18		"""
19
20	1	MAX_LENGTH_VARCHAR = 4096
21		"""
22		Maximum length of a varchar.
23		"""
24
25	1	MAX_LENGTH_BINARY = 255
26		"""
27		Maximum length of a varbinary.
28		"""
29
30	1	MAX_LENGTH_VARBINARY = 4096
31		"""
32		Maximum length of a varbinary.
33		"""
34
35		# ------------------------------------------------------------------------------------------------------------------
36	1	def __init__(self, io: StratumStyle, config: ConfigParser):
37		"""
38		Object constructor.
39
40		:param PyStratumStyle io: The output decorator.
41		"""
42		MySqlWorker.__init__(self, io, config)
43		CommonRoutineLoaderWorker.__init__(self, io, config)
44
45		self._character_set_client: Optional[str] = None
46		"""
47		The default character set under which the stored routine will be loaded and run.
48		"""
49
50		self._collation_connection: Optional[str] = None
51		"""
52		The default collate under which the stored routine will be loaded and run.
53		"""
54
55		self._sql_mode: Optional[str] = None
56		"""
57
58		"""
59
60		# ------------------------------------------------------------------------------------------------------------------
61	1	def __save_column_types_exact(self, rows: List[Dict[str, Any]]) -> None:
62		"""
63		Saves the exact column types as replace pairs.
64
65		:param list[dict[str,*]] rows: The columns types.
66		"""
67		for row in rows:
68		key = row['table_name'] + '.' + row['column_name'] + '%type'
69
70		value = row['column_type']
71		if row['character_set_name']:
72		value += ' character set ' + row['character_set_name']
73
74		self._add_replace_pair(key, value, False)
75
76		# ------------------------------------------------------------------------------------------------------------------
77	1	def __save_column_types_max_length(self, rows: List[Dict[str, Any]]) -> None:
78		"""
79		Saves the column types with maximum length as replace pairs.
80
81		:param list[dict[str,*]] rows: The columns types.
82		"""
83		for row in rows:
84		key = row['table_name'] + '.' + row['column_name'] + '%max-type'
85
86		if row['data_type'] == 'char':
87		value = row['data_type'] + '(' + str(self.MAX_LENGTH_CHAR) + ')'
88		value += ' character set ' + row['character_set_name']
89		self._add_replace_pair(key, value, False)
90
91		if row['data_type'] == 'varchar':
92		value = row['data_type'] + '(' + str(self.MAX_LENGTH_VARCHAR) + ')'
93		value += ' character set ' + row['character_set_name']
94		self._add_replace_pair(key, value, False)
95
96		elif row['data_type'] == 'binary':
97		value = row['data_type'] + '(' + str(self.MAX_LENGTH_BINARY) + ')'
98		self._add_replace_pair(key, value, False)
99
100		elif row['data_type'] == 'varbinary':
101		value = row['data_type'] + '(' + str(self.MAX_LENGTH_VARBINARY) + ')'
102		self._add_replace_pair(key, value, False)
103
104		# ------------------------------------------------------------------------------------------------------------------
105	1	def _get_column_type(self) -> None:
106		"""
107		Selects schema, table, column names and the column type from MySQL and saves them as replace pairs.
108		"""
109		rows = self._dl.get_all_table_columns()
110		self.__save_column_types_exact(rows)
111		self.__save_column_types_max_length(rows)
112
113		self._io.text('Selected {0} column types for substitution'.format(len(rows)))
114
115		# ------------------------------------------------------------------------------------------------------------------
116	1	def create_routine_loader_helper(self,
117		routine_name: str,
118		pystratum_old_metadata: Optional[Dict],
119		rdbms_old_metadata: Optional[Dict]) -> MySqlRoutineLoaderHelper:
120		"""
121		Creates a Routine Loader Helper object.
122
123		:param str routine_name: The name of the routine.
124		:param dict pystratum_old_metadata: The old metadata of the stored routine from PyStratum.
125		:param dict rdbms_old_metadata: The old metadata of the stored routine from MySQL.
126
127		:rtype: MySqlRoutineLoaderHelper
128		"""
129		return MySqlRoutineLoaderHelper(self._io,
130		self._dl,
131		self._source_file_names[routine_name],
132		self._source_file_encoding,
133		pystratum_old_metadata,
134		self._replace_pairs,
135		rdbms_old_metadata,
136		self._sql_mode,
137		self._character_set_client,
138		self._collation_connection)
139
140		# ------------------------------------------------------------------------------------------------------------------
141	1	def _get_old_stored_routine_info(self) -> None:
142		"""
143		Retrieves information about all stored routines in the current schema.
144		"""
145		rows = self._dl.get_routines()
146		self._rdbms_old_metadata = {}
147		for row in rows:
148		self._rdbms_old_metadata[row['routine_name']] = row
149
150		# ------------------------------------------------------------------------------------------------------------------
151	1	def _get_correct_sql_mode(self) -> None:
152		"""
153		Gets the SQL mode in the order as preferred by MySQL.
154		"""
155		self._sql_mode = self._dl.get_correct_sql_mode(self._sql_mode)
156
157		# ------------------------------------------------------------------------------------------------------------------
158	1	def _drop_obsolete_routines(self) -> None:
159		"""
160		Drops obsolete stored routines (i.e. stored routines that exits in the current schema but for
161		which we don't have a source file).
162		"""
163		for routine_name, values in self._rdbms_old_metadata.items():
164		if routine_name not in self._source_file_names:
165		self._io.text("Dropping {0} <dbo>{1}</dbo>".format(values['routine_type'].lower(), routine_name))
166		self._dl.drop_stored_routine(values['routine_type'], routine_name)
167
168		# ------------------------------------------------------------------------------------------------------------------
169	1	def _read_configuration_file(self) -> None:
170		"""
171		Reads parameters from the configuration file.
172		"""
173		CommonRoutineLoaderWorker._read_configuration_file(self)
174
175		self._character_set_client = self._config.get('database', 'character_set_client', fallback='utf-8')
176		self._collation_connection = self._config.get('database', 'collation_connection', fallback='utf8_general_ci')
177		self._sql_mode = self._config.get('database', 'sql_mode')
178
179		# ----------------------------------------------------------------------------------------------------------------------
180

DatabaseStratum / py-stratum-mysql

Push — master ( 7c9493...8a6195 )

MySqlRoutineLoaderWorker.__init__() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

MySqlRoutineLoaderWorker.init() A