Coverage for pystratum_common/wrapper/RowsWithKeyWrapper.py: 0%
53 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
2from typing import Any, Dict
4from pystratum_common.wrapper.Wrapper import Wrapper
7class RowsWithKeyWrapper(Wrapper):
8 """
9 Parent class wrapper method generator for stored procedures whose result set must be returned using tree
10 structure using a combination of unique columns.
11 """
13 # ------------------------------------------------------------------------------------------------------------------
14 def _return_type_hint(self) -> str:
15 """
16 Returns the return type hint of the wrapper method.
17 """
18 return 'Dict'
20 # ------------------------------------------------------------------------------------------------------------------
21 def _get_docstring_return_type(self):
22 """
23 Returns the return type of the wrapper methods to be used in the docstring.
24 """
25 return 'dict'
27 # ------------------------------------------------------------------------------------------------------------------
28 @abc.abstractmethod
29 def _write_execute_rows(self, routine: Dict[str, Any]) -> None:
30 raise NotImplementedError()
32 # ------------------------------------------------------------------------------------------------------------------
33 def _write_result_handler(self, routine: Dict[str, Any]) -> None:
34 """
35 Generates code for calling the stored routine in the wrapper method.
36 """
37 self._write_line('ret = {}')
38 self._write_execute_rows(routine)
39 self._write_line('for row in rows:')
41 num_of_dict = len(routine['columns'])
43 i = 0
44 while i < num_of_dict:
45 value = "row['{0!s}']".format(routine['columns'][i])
47 stack = ''
48 j = 0
49 while j < i:
50 stack += "[row['{0!s}']]".format(routine['columns'][j])
51 j += 1
52 line = 'if {0!s} in ret{1!s}:'.format(value, stack)
53 self._write_line(line)
54 i += 1
56 line = "raise Exception('Duplicate key for %s.' % str(({0!s})))". \
57 format(", ".join(["row['{0!s}']".format(column_name) for column_name in routine['columns']]))
59 self._write_line(line)
60 self._indent_level_down()
62 i = num_of_dict
63 while i > 0:
64 self._write_line('else:')
66 part1 = ''
67 j = 0
68 while j < i - 1:
69 part1 += "[row['{0!s}']]".format(routine['columns'][j])
70 j += 1
71 part1 += "[row['{0!s}']]".format(routine['columns'][j])
73 part2 = ''
74 j = i - 1
75 while j < num_of_dict:
76 if j + 1 != i:
77 part2 += "{{row['{0!s}']: ".format(routine['columns'][j])
78 j += 1
79 part2 += "row" + ('}' * (num_of_dict - i))
81 line = "ret{0!s} = {1!s}".format(part1, part2)
82 self._write_line(line)
83 self._indent_level_down()
84 if i > 1:
85 self._indent_level_down()
86 i -= 1
88 self._write_line()
89 self._write_line('return ret')
91# ----------------------------------------------------------------------------------------------------------------------