Coverage for postrfp/model/questionnaire/answering.py: 98%

1from enum import Enum 

2from datetime import datetime 

3from typing import Optional, TYPE_CHECKING 

4 

5from sqlalchemy import ForeignKey, Index, Integer, Unicode, text, DateTime 

6from sqlalchemy.orm import Mapped, mapped_column, relationship 

7from sqlalchemy.types import INTEGER, TEXT, TypeDecorator, SMALLINT, VARCHAR 

8 

9from postrfp.model.meta import AttachmentMixin, Base 

10 

11if TYPE_CHECKING: 

12 from postrfp.model.issue import Issue 

13 from postrfp.model.project import Project 

14 from postrfp.model.questionnaire.nodes import QuestionInstance, QElement 

15 

16 

17class AnswerReport(Base): 

18 __tablename__ = "answer_reports" 

19 

20 id: Mapped[int] = mapped_column(Integer, primary_key=True) 

21 project_id: Mapped[int] = mapped_column( 

22 ForeignKey("projects.id", ondelete="CASCADE"), 

23 nullable=False, 

24 index=True, 

25 ) 

26 title: Mapped[str] = mapped_column(Unicode(100), nullable=False) 

27 definition: Mapped[str] = mapped_column(TEXT(), nullable=False) 

28 project: Mapped["Project"] = relationship( 

29 "Project", back_populates="answer_reports" 

30 ) 

31 

32 def __repr__(self): 

33 return f"<AnswerReport {self.id} - {self.title}>" 

34 

35 

36class ResponseStatus(Enum): 

37 NOT_ANSWERED = 0 

38 ANSWERED = 10 

39 FOR_REVIEW = 20 

40 REJECTED = 30 

41 APPROVED = 40 

42 

43 

44class ResponseStatusCol(TypeDecorator): 

45 """ 

46 A custom SQLAlchemy type that maps database integer 

47 values to ResponseStatus Enum values 

48 """ 

49 

50 impl = SMALLINT() 

51 

52 cache_ok = True 

53 

54 def process_bind_param(self, response_status_enum, dialect): 

55 return response_status_enum.value 

56 

57 def process_result_value(self, int_value, dialect): 

58 return ResponseStatus(int_value) 

59 

60 

61class QuestionResponseState(Base): 

62 __tablename__ = "question_response_states" 

63 __table_args__ = ( 

64 Index("unique_resp_state", "issue_id", "question_instance_id", unique=True), 

65 ) 

66 public_attrs = [ 

67 "status", 

68 "allocated_by", 

69 "allocated_to", 

70 "approved_by", 

71 "updated_by", 

72 "date_updated", 

73 ] 

74 

75 id: Mapped[int] = mapped_column(Integer, primary_key=True) 

76 

77 issue_id: Mapped[int] = mapped_column( 

78 ForeignKey("issues.id", ondelete="CASCADE"), 

79 nullable=False, 

80 index=True, 

81 server_default=text("'0'"), 

82 ) 

83 

84 question_instance_id: Mapped[int] = mapped_column( 

85 ForeignKey("question_instances.id", ondelete="CASCADE"), 

86 nullable=False, 

87 index=True, 

88 server_default=text("'0'"), 

89 ) 

90 

91 status: Mapped[ResponseStatus] = mapped_column( 

92 ResponseStatusCol, nullable=False, server_default=text("'0'") 

93 ) 

94 

95 allocated_by: Mapped[Optional[str]] = mapped_column( 

96 VARCHAR(length=50), ForeignKey("users.id", ondelete="SET NULL") 

97 ) 

98 allocated_to: Mapped[Optional[str]] = mapped_column( 

99 VARCHAR(length=50), ForeignKey("users.id", ondelete="SET NULL") 

100 ) 

101 approved_by: Mapped[Optional[str]] = mapped_column( 

102 VARCHAR(length=50), ForeignKey("users.id", ondelete="SET NULL") 

103 ) 

104 updated_by: Mapped[Optional[str]] = mapped_column( 

105 VARCHAR(length=50), ForeignKey("users.id", ondelete="SET NULL") 

106 ) 

107 

108 date_updated: Mapped[Optional[datetime]] = mapped_column(DateTime) 

109 

110 issue: Mapped["Issue"] = relationship( 

111 "Issue", 

112 back_populates="response_states", 

113 ) 

114 question_instance: Mapped["QuestionInstance"] = relationship("QuestionInstance") 

115 

116 

117class Answer(Base): 

118 __tablename__ = "answers" 

119 __table_args__ = ( 

120 Index( 

121 "unique_issue_question_element_answer", 

122 "issue_id", 

123 "question_instance_id", 

124 "element_id", 

125 unique=True, 

126 ), 

127 Index("ft_answer", "answer", mariadb_prefix="FULLTEXT"), 

128 ) 

129 

130 issue_id: Mapped[int] = mapped_column( 

131 Integer, 

132 ForeignKey("issues.id", ondelete="CASCADE"), # Removed name="answers_ibfk_1" 

133 nullable=False, 

134 default=0, 

135 index=True, 

136 ) 

137 

138 answer: Mapped[str] = mapped_column(TEXT(), nullable=True) 

139 

140 element_id: Mapped[int] = mapped_column( 

141 Integer, 

142 ForeignKey("question_elements.id"), # Removed name="answers_ibfk_2" 

143 nullable=False, 

144 index=True, 

145 server_default=text("'0'"), 

146 ) 

147 autoscore: Mapped[Optional[int]] = mapped_column(INTEGER) 

148 question_instance_id: Mapped[int] = mapped_column( 

149 Integer, 

150 ForeignKey( 

151 "question_instances.id", ondelete="CASCADE" 

152 ), # Removed name="answers_ibfk_3" 

153 nullable=False, 

154 index=True, 

155 ) 

156 

157 issue: Mapped["Issue"] = relationship("Issue", back_populates="answers") 

158 

159 element: Mapped["QElement"] = relationship("QElement", back_populates="answers") 

160 

161 question_instance: Mapped["QuestionInstance"] = relationship( 

162 "QuestionInstance", back_populates="_answers", uselist=False 

163 ) 

164 attachment: Mapped["AAttachment"] = relationship( 

165 "AAttachment", 

166 back_populates="answer", 

167 passive_deletes=True, 

168 uselist=False, 

169 ) 

170 

171 def as_dict(self): 

172 return {"answer_id": self.id, "issue_id": self.issue_id, "answer": self.answer} 

173 

174 def __repr__(self): 

175 return f"Answer ID {self.id}, Issue: {self.issue_id}, QuestionInstance: {self.question_instance_id}" 

176 

177 

178class AAttachment(AttachmentMixin, Base): 

179 __tablename__ = "answer_attachments" 

180 

181 public_attrs = ( 

182 "id,size,filename,url,respondent,question_number,question_id" 

183 ).split(",") 

184 answer_id: Mapped[Optional[int]] = mapped_column( 

185 Integer, ForeignKey("answers.id", ondelete="SET NULL"), unique=True 

186 ) 

187 

188 answer: Mapped[Optional["Answer"]] = relationship( 

189 Answer, 

190 uselist=False, 

191 back_populates="attachment", 

192 ) 

193 

194 def __repr__(self) -> str: 

195 return f"<AAttachment #{self.id} - {self.filename}>"