Coverage for pystratum_common/backend/CommonRoutineWrapperGeneratorWorker.py: 0%
79 statements
« prev ^ index » next coverage.py v7.5.1, created at 2024-05-13 08:46 +0200
« prev ^ index » next coverage.py v7.5.1, created at 2024-05-13 08:46 +0200
1import abc
2import configparser
3import json
4import os
5from typing import Any, Dict, Optional
7from pystratum_backend.RoutineWrapperGeneratorWorker import RoutineWrapperGeneratorWorker
8from pystratum_backend.StratumIO import StratumIO
9from pystratum_common.Util import Util
12class CommonRoutineWrapperGeneratorWorker(RoutineWrapperGeneratorWorker):
13 """
14 Class for generating a class with wrapper methods for calling stored routines in a MySQL database.
15 """
17 # ------------------------------------------------------------------------------------------------------------------
18 def __init__(self, io: StratumIO, config: configparser.ConfigParser):
19 """
20 Object constructor.
22 :param io: The output decorator.
23 """
24 self._code: str = ''
25 """
26 The generated Python code buffer.
27 """
29 self._lob_as_string_flag: bool = False
30 """
31 If true BLOBs and CLOBs must be treated as strings.
32 """
34 self._metadata_filename: Optional[str] = None
35 """
36 The filename of the file with the metadata of all stored procedures.
37 """
39 self._parent_class_name: Optional[str] = None
40 """
41 The class name of the parent class of the routine wrapper.
42 """
44 self._parent_class_namespace: Optional[str] = None
45 """
46 The namespace of the parent class of the routine wrapper.
47 """
49 self._wrapper_class_name: Optional[str] = None
50 """
51 The class name of the routine wrapper.
52 """
54 self._wrapper_filename: Optional[str] = None
55 """
56 The filename where the generated wrapper class must be stored.
57 """
59 self._io: StratumIO = io
60 """
61 The output decorator.
62 """
64 self._config = config
65 """
66 The configuration object.
67 """
69 # ------------------------------------------------------------------------------------------------------------------
70 def execute(self) -> int:
71 """
72 The "main" of the wrapper generator. Returns 0 on success, 1 if one or more errors occurred.
73 """
74 self._read_configuration_file()
76 if self._wrapper_class_name:
77 self._io.title('Wrapper')
79 self.__generate_wrapper_class()
81 self._io.write_line('')
82 else:
83 self._io.log_verbose('Wrapper not enabled')
85 return 0
87 # ------------------------------------------------------------------------------------------------------------------
88 def __generate_wrapper_class(self) -> None:
89 """
90 Generates the wrapper class.
91 """
92 routines = self._read_routine_metadata()
94 self._write_class_header()
96 if routines:
97 for routine_name in sorted(routines):
98 if routines[routine_name]['designation'] != 'hidden':
99 self._write_routine_function(routines[routine_name])
100 else:
101 self._io.error('No files with stored routines found')
103 self._write_class_trailer()
105 Util.write_two_phases(self._wrapper_filename, self._code, self._io)
107 # ------------------------------------------------------------------------------------------------------------------
108 def _read_configuration_file(self) -> None:
109 """
110 Reads parameters from the configuration file.
111 """
112 self._parent_class_name = self._config.get('wrapper', 'parent_class')
113 self._parent_class_namespace = self._config.get('wrapper', 'parent_class_namespace')
114 self._wrapper_class_name = self._config.get('wrapper', 'wrapper_class')
115 self._wrapper_filename = self._config.get('wrapper', 'wrapper_file')
116 self._metadata_filename = self._config.get('wrapper', 'metadata')
117 self._lob_as_string_flag = bool(self._config.get('wrapper', 'lob_as_string'))
119 # ------------------------------------------------------------------------------------------------------------------
120 def _read_routine_metadata(self) -> Dict:
121 """
122 Returns the metadata of stored routines.
123 """
124 metadata = {}
125 if os.path.isfile(self._metadata_filename):
126 with open(self._metadata_filename, 'r') as file:
127 metadata = json.load(file)
129 return metadata
131 # ------------------------------------------------------------------------------------------------------------------
132 def _write_class_header(self) -> None:
133 """
134 Generate a class header for stored routine wrapper.
135 """
136 self._write_line('from typing import Any, Dict, List, Optional, Union')
137 self._write_line()
138 self._write_line('from {0!s} import {1!s}'.format(self._parent_class_namespace, self._parent_class_name))
139 self._write_line()
140 self._write_line()
141 self._write_line('class {0!s}({1!s}):'.format(self._wrapper_class_name, self._parent_class_name))
142 self._write_line(' """')
143 self._write_line(' The stored routines wrappers.')
144 self._write_line(' """')
146 # ------------------------------------------------------------------------------------------------------------------
147 def _write_line(self, text: str = '') -> None:
148 """
149 Writes a line with Python code to the generate code buffer.
151 :param text: The line with Python code.
152 """
153 if text:
154 self._code += str(text) + "\n"
155 else:
156 self._code += "\n"
158 # ------------------------------------------------------------------------------------------------------------------
159 def _write_class_trailer(self) -> None:
160 """
161 Generate a class trailer for stored routine wrapper.
162 """
163 self._write_line()
164 self._write_line()
165 self._write_line('# ' + ('-' * 118))
167 # ------------------------------------------------------------------------------------------------------------------
168 @abc.abstractmethod
169 def _write_routine_function(self, routine: Dict[str, Any]) -> None:
170 """
171 Generates a complete wrapper method for a stored routine.
173 :param routine: The metadata of the stored routine.
174 """
175 raise NotImplementedError()
177# ----------------------------------------------------------------------------------------------------------------------