Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

1import abc 

2from typing import Any, Dict 

3 

4from pystratum_common.wrapper.Wrapper import Wrapper 

5 

6 

7class RowsWithKeyWrapper(Wrapper): 

8 """ 

9 Parent class wrapper wrapper method generator for stored procedures whose result set must be returned using tree 

10 structure using a combination of unique columns. 

11 """ 

12 

13 # ------------------------------------------------------------------------------------------------------------------ 

14 def _return_type_hint(self) -> str: 

15 """ 

16 Returns the return type hint of the wrapper method. 

17 

18 :rtype: str 

19 """ 

20 return 'Dict' 

21 

22 # ------------------------------------------------------------------------------------------------------------------ 

23 def _get_docstring_return_type(self): 

24 """ 

25 Returns the return type of the wrapper methods the be used in the docstring. 

26 

27 :rtype: str 

28 """ 

29 return 'dict' 

30 

31 # ------------------------------------------------------------------------------------------------------------------ 

32 @abc.abstractmethod 

33 def _write_execute_rows(self, routine: Dict[str, Any]) -> None: 

34 raise NotImplementedError() 

35 

36 # ------------------------------------------------------------------------------------------------------------------ 

37 def _write_result_handler(self, routine: Dict[str, Any]) -> None: 

38 """ 

39 Generates code for calling the stored routine in the wrapper method. 

40 """ 

41 self._write_line('ret = {}') 

42 self._write_execute_rows(routine) 

43 self._write_line('for row in rows:') 

44 

45 num_of_dict = len(routine['columns']) 

46 

47 i = 0 

48 while i < num_of_dict: 

49 value = "row['{0!s}']".format(routine['columns'][i]) 

50 

51 stack = '' 

52 j = 0 

53 while j < i: 

54 stack += "[row['{0!s}']]".format(routine['columns'][j]) 

55 j += 1 

56 line = 'if {0!s} in ret{1!s}:'.format(value, stack) 

57 self._write_line(line) 

58 i += 1 

59 

60 line = "raise Exception('Duplicate key for %s.' % str(({0!s})))". \ 

61 format(", ".join(["row['{0!s}']".format(column_name) for column_name in routine['columns']])) 

62 

63 self._write_line(line) 

64 self._indent_level_down() 

65 

66 i = num_of_dict 

67 while i > 0: 

68 self._write_line('else:') 

69 

70 part1 = '' 

71 j = 0 

72 while j < i - 1: 

73 part1 += "[row['{0!s}']]".format(routine['columns'][j]) 

74 j += 1 

75 part1 += "[row['{0!s}']]".format(routine['columns'][j]) 

76 

77 part2 = '' 

78 j = i - 1 

79 while j < num_of_dict: 

80 if j + 1 != i: 

81 part2 += "{{row['{0!s}']: ".format(routine['columns'][j]) 

82 j += 1 

83 part2 += "row" + ('}' * (num_of_dict - i)) 

84 

85 line = "ret{0!s} = {1!s}".format(part1, part2) 

86 self._write_line(line) 

87 self._indent_level_down() 

88 if i > 1: 

89 self._indent_level_down() 

90 i -= 1 

91 

92 self._write_line() 

93 self._write_line('return ret') 

94 

95# ----------------------------------------------------------------------------------------------------------------------