Coverage for postrfp/buyer/api/io/excel_import.py: 98%

1# Description: Import questions from an Excel file into a project 

2from openpyxl import load_workbook 

3from openpyxl.worksheet.worksheet import Worksheet 

4from sqlalchemy.orm import object_session 

5 

6from postrfp.model import Project 

7from postrfp.model import ( 

8 Section, 

9 RadioChoices, 

10 TextInput, 

11 Label, 

12 QuestionInstance, 

13 QuestionDefinition, 

14) 

15 

16 

17class XLCell: 

18 def __init__(self, col_idx: int): 

19 self.col_idx = col_idx 

20 

21 def __get__(self, obj: "QRow", type=None) -> object: 

22 return obj.row[self.col_idx].value 

23 

24 

25class QRow: 

26 section_title = XLCell(0) 

27 question_title = XLCell(1) 

28 question_text = XLCell(2) 

29 multi_choices = XLCell(3) 

30 comments_title = XLCell(4) 

31 

32 def __init__(self, row: tuple): 

33 self.row = row 

34 

35 def is_root_question(self): 

36 return not self.section_title 

37 

38 def choices_list(self): 

39 cd = [] 

40 for choice_str in self.multi_choices.split(";"): 

41 choice, autoscore = RadioChoices.split_choice_string(choice_str) 

42 cd.append({"label": choice, "autoscore": autoscore}) 

43 if len(cd) < 2: 

44 raise ValueError( 

45 "At least two values are required for multiple choice options" 

46 ) 

47 return cd 

48 

49 

50def row_to_question(qrow) -> QuestionDefinition: 

51 qdef = QuestionDefinition(title=qrow.question_title) 

52 elems = qdef.elements 

53 elems.append(Label(label=qrow.question_text, row=1, col=1)) 

54 if qrow.multi_choices: 

55 choices = RadioChoices(choices=qrow.choices_list(), row=2, col=1) 

56 elems.append(choices) 

57 if qrow.comments_title: 

58 elems.append(Label(label=qrow.comments_title, row=3, col=1)) 

59 elems.append(TextInput(height=5, width=60, row=4, col=1)) 

60 else: 

61 if qrow.comments_title: 

62 elems.append(Label(label=qrow.comments_title, row=2, col=1)) 

63 elems.append(TextInput(height=5, width=60, row=3, col=1)) 

64 else: 

65 elems.append(TextInput(height=5, width=60, row=2, col=1)) 

66 return qdef 

67 

68 

69class ExcelQImporter: 

70 """ 

71 Import questions from an Excel file into a project. 

72 

73 The Excel file must have the following columns: 

74 1. Section Title 

75 2. Question Title 

76 3. Question Text 

77 4. Multiple Choices 

78 5. Comments Title 

79 

80 The Multiple Choices column is a semicolon-separated list of choices. 

81 Each choice can have an optional autoscore value separated by a colon. 

82 For example: 

83 "Yes:1;No:0;Don't know:0" 

84 

85 """ 

86 

87 def __init__(self, parent_section: Section) -> None: 

88 self.created_count = 0 

89 if parent_section is None: 

90 raise ValueError("missing required Section object, parent_section") 

91 self.parent_section = parent_section 

92 self.project: Project = parent_section.project 

93 self.created_sections: dict[object, Section] = {} 

94 self.session = object_session(parent_section) 

95 if self.session is None: 

96 raise ValueError("parent_section must be attached to a session") 

97 

98 def read_questions_excel(self, filename): 

99 wb = load_workbook(filename=filename, read_only=True) 

100 try: 

101 self.process_rows(wb["Sheet1"]) 

102 finally: 

103 wb.close() 

104 

105 def process_rows(self, ws: Worksheet): 

106 # max_row is 501 to allow a maximum of 500 rows since 

107 # rows are indexed from 1, not zero, and there is a header row 

108 for row in ws.iter_rows(min_row=2, max_row=501): 

109 qr = QRow(row) 

110 if qr.question_title: 

111 self.process_one_row(qr) 

112 self.created_count += 1 

113 

114 def process_one_row(self, qrow: QRow): 

115 qdef = row_to_question(qrow) 

116 assert self.session is not None 

117 

118 self.session.add(qdef) 

119 self.session.flush() # Assign qdef an ID 

120 

121 sec: Section 

122 if qrow.is_root_question(): 

123 sec = self.parent_section 

124 elif qrow.section_title in self.created_sections: 

125 sec = self.created_sections[qrow.section_title] 

126 else: 

127 sec = Section(title=qrow.section_title, project_id=self.project.id) 

128 assert self.session is not None 

129 self.session.add(sec) 

130 self.session.flush() # Assign section an ID 

131 

132 with self.session.no_autoflush: 

133 self.parent_section.subsections.append(sec) 

134 

135 self.created_sections[qrow.section_title] = sec 

136 

137 qi = QuestionInstance( 

138 question_def_id=qdef.id, project_id=self.project.id, section_id=sec.id 

139 ) 

140 

141 self.session.add(qi) 

142 

143 # Explicitly establish the relationships after adding to session 

144 with self.session.no_autoflush: 

145 sec.questions.append(qi)