Coverage for postrfp/shared/serial/qmodels.py: 100%

1from datetime import datetime 

2from enum import Enum 

3from typing import List, Union, Literal, Optional 

4 

5 

6from pydantic import ( 

7 field_validator, 

8 StringConstraints, 

9 ConfigDict, 

10 BaseModel, 

11 Field, 

12 RootModel, 

13) 

14from typing_extensions import Annotated 

15 

16 

17class ElTypes(str, Enum): 

18 TX = "TX" 

19 LB = "LB" 

20 CR = "CR" 

21 CC = "CC" 

22 CB = "CB" 

23 QA = "QA" 

24 AT = "AT" 

25 

26 

27class IntIdModel(BaseModel): 

28 id: Optional[int] = Field(default=None, json_schema_extra={"readOnly": True}) 

29 

30 

31class Choice(BaseModel): 

32 model_config = ConfigDict(extra="forbid") 

33 

34 autoscore: Optional[Annotated[int, Field(ge=0, le=1000)]] = None 

35 label: Annotated[str, StringConstraints(min_length=1, max_length=1024)] 

36 

37 

38class QElement(IntIdModel): 

39 model_config = ConfigDict(from_attributes=True, extra="forbid") 

40 

41 el_type: Literal["TX", "LB", "CR", "CC", "CB", "QA", "AT"] 

42 label: Optional[Annotated[str, StringConstraints(max_length=4086)]] = None 

43 colspan: Annotated[int, Field(ge=1, le=10)] = 1 

44 rowspan: Annotated[int, Field(ge=1, le=50)] = 1 

45 

46 

47class Checkbox(QElement): 

48 el_type: Literal["CB"] 

49 

50 

51class Label(QElement): 

52 el_type: Literal["LB"] 

53 

54 

55class TextInput(QElement): 

56 el_type: Literal["TX"] 

57 height: Annotated[int, Field(gt=0, lt=40)] = 1 

58 width: Annotated[int, Field(gt=0, lt=500)] = 1 

59 regexp: Optional[str] = None 

60 mandatory: bool = False 

61 

62 

63class MultiChoice(BaseModel): 

64 choices: Annotated[List[Choice], Field(min_length=2, max_length=20)] 

65 

66 

67class SelectDropdown(QElement, MultiChoice): 

68 el_type: Literal["CC"] 

69 mandatory: bool = False 

70 

71 

72class RadioChoices(QElement, MultiChoice): 

73 el_type: Literal["CR"] 

74 mandatory: bool = False 

75 

76 

77class QuestionAttachment(QElement): 

78 el_type: Literal["QA"] 

79 

80 

81class UploadField(QElement): 

82 el_type: Literal["AT"] 

83 mandatory: bool = False 

84 

85 

86el_types = { 

87 ElTypes.TX: TextInput, 

88 ElTypes.LB: Label, 

89 ElTypes.CR: RadioChoices, 

90 ElTypes.CC: SelectDropdown, 

91 ElTypes.CB: Checkbox, 

92 ElTypes.QA: QuestionAttachment, 

93 ElTypes.AT: UploadField, 

94} 

95 

96 

97class ElRow(RootModel): 

98 root: List[ 

99 Union[ 

100 TextInput, 

101 Label, 

102 RadioChoices, 

103 SelectDropdown, 

104 Checkbox, 

105 QuestionAttachment, 

106 UploadField, 

107 ] 

108 ] = Field(..., max_length=10, min_length=0) 

109 

110 

111class ElGrid(RootModel): 

112 root: List[ElRow] = Field(..., max_length=50, min_length=0) 

113 

114 

115class QuestionDef(BaseModel): 

116 title: Annotated[str, StringConstraints(min_length=2, max_length=1024)] 

117 elements: ElGrid 

118 model_config = ConfigDict(from_attributes=True) 

119 

120 @field_validator("elements", mode="before") 

121 @classmethod 

122 def nest_elements(cls, els): 

123 """ 

124 Manage element nesting & row/col values (a list of lists) 

125 

126 If the data being validated is already nested then assign 

127 row/col values according to the position of the element in the 

128 list of lists (2D array). 

129 

130 If we are validating an ORM model then build up the list of lists 

131 

132 Element.row an Element.col attributes are not exposed to the outside 

133 world. For them to be set on incoming data it's necessary that the 

134 pydantic QElement model config allows extra attributes 

135 """ 

136 if els is None: 

137 raise ValueError("At least one row is required") 

138 if len(els) == 0: 

139 return els 

140 if isinstance(els[0], list): 

141 # We are validating a JSON dict were elements are 

142 # already arranged in rows/columns, i.e. incoming json 

143 for row_idx, row in enumerate(els, start=1): 

144 for col_idx, el in enumerate(row, start=1): 

145 # el['row'] = row_idx 

146 # el['col'] = col_idx 

147 el_cls = el_types[el["el_type"]] 

148 el_model = el_cls(**el) 

149 els[row_idx - 1][col_idx - 1] = el_model 

150 # validate_rowspans(els) 

151 return els 

152 

153 # We are validating an ORM model - elements is a list of model objects 

154 rows = [[]] 

155 current_row = rows[0] 

156 current_row_idx = 1 

157 

158 if els[0].row != 1: 

159 el = els[0] 

160 raise ValueError( 

161 f"{el.el_type} element {el.id} row is {el.row}, should be 1" 

162 ) 

163 

164 for el in els: 

165 mod_cls = el_types[el.el_type] 

166 mod = mod_cls.model_validate(el) 

167 if el.row == current_row_idx: 

168 current_row.append(mod) 

169 elif el.row == current_row_idx + 1: 

170 current_row = [mod] 

171 rows.append(current_row) 

172 current_row_idx = el.row 

173 else: 

174 raise ValueError(f"row {el.row} invalid for el {el.id}") 

175 return rows 

176 

177 

178class Question(IntIdModel, QuestionDef): 

179 number: str 

180 section_id: Annotated[int, Field(lt=4294967295)] | None = None 

181 

182 

183class RespondentAnswer(BaseModel): 

184 model_config = ConfigDict(from_attributes=True) 

185 answer_id: int 

186 answer: str 

187 respondent_id: str 

188 issue_id: int 

189 project_id: int 

190 project_title: str 

191 date_published: Optional[datetime] 

192 

193 

194class RespondentAnswers(RootModel): 

195 root: List[RespondentAnswer] 

196 

197 

198class ExcelImportResult(BaseModel): 

199 imported_count: int = Field( 

200 ..., description="Count of the number of imported questions" 

201 )