Coverage for postrfp/model/composite.py: 100%

1import logging 

2from hashlib import md5 

3from typing import Optional 

4 

5from sqlalchemy import types, ForeignKey, select, Index 

6from sqlalchemy.orm import relationship, Session, Mapped, mapped_column 

7from sqlalchemy.dialects.mysql import insert 

8 

9from .meta import Base 

10 

11 

12from postrfp.model.questionnaire.nodes import QuestionDefinition, QuestionInstance 

13 

14 

15log = logging.getLogger(__name__) 

16 

17 

18class QuestionMeta(Base): 

19 __tablename__ = "questions_meta" 

20 __table_args__ = ( 

21 Index( 

22 "ft_text_meta", "alltext", mariadb_prefix="FULLTEXT", postgresql_using="gin" 

23 ), 

24 ) 

25 

26 question_id: Mapped[int] = mapped_column( 

27 types.Integer, 

28 ForeignKey("questions.id", ondelete="CASCADE"), 

29 unique=True, 

30 nullable=False, # Ensure question_id cannot be NULL 

31 ) 

32 question_def: Mapped["QuestionDefinition"] = relationship( 

33 "QuestionDefinition", 

34 back_populates="meta", 

35 viewonly=True, 

36 ) 

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

38 "QuestionInstance", 

39 secondary="questions", 

40 back_populates="meta", 

41 viewonly=True, 

42 ) 

43 

44 # For efficient & simplified full text querying of questions 

45 alltext: Mapped[Optional[str]] = mapped_column(types.TEXT, nullable=True) 

46 

47 # Identify the structure of the input elements for uniform reporting 

48 # similar to the yesno reports developed for T2R 

49 # Support for 'questions like this' functionality 

50 signature: Mapped[Optional[str]] = mapped_column( 

51 types.VARCHAR(length=64), index=True, nullable=True 

52 ) 

53 

54 def __repr__(self) -> str: 

55 return f"<QuestionMeta, QDef # {self.question_id}, signature: {self.signature}>" 

56 

57 

58def qsignature(sig_list: list[str]) -> str: 

59 # The following hash is not used in any security context. It is only used 

60 # to generate unique values, collisions are acceptable so md5 is 

61 # sufficient. "nosec" is used to disable bandit warning for this hash. 

62 return md5("|".join(sig_list).encode("utf8")).hexdigest() # nosec 

63 

64 

65def update_meta_row_stmt(q): 

66 qtext_set = [q.title] 

67 sig_list = [] 

68 for el in q.elements: 

69 sig_list.append(f"{el.el_type}-{el.row}-{el.col}") 

70 if el.el_type in ("LB", "CB") and el.label is not None: 

71 qtext_set.append(el.label) 

72 if el.choices: 

73 for c in el.choices: 

74 if "label" in c and c["label"] is not None: 

75 qtext_set.append(c["label"]) 

76 qtext = " ".join(qtext_set) 

77 sig = qsignature(sig_list) 

78 stmt = insert(QuestionMeta).values(question_id=q.id, alltext=qtext, signature=sig) 

79 return stmt.on_duplicate_key_update(alltext=qtext, signature=sig) 

80 

81 

82def update_qmeta_table(session: Session, qdef_ids): 

83 """ 

84 Update the questions_meta database table for the given QuestionDefintion ids. This function 

85 should be called whenever a question defintion is inserted or updated. 

86 

87 Each row of the questions_meta table maps to one QuestionDefinition and provides a FullText 

88 indexed column for easy searching and a regularly indexed "signature" column to allow 

89 questions with the identical structures to be identified for uniform reporting 

90 """ 

91 conn = session.connection() 

92 stmt = select(QuestionDefinition).where(QuestionDefinition.id.in_(qdef_ids)) 

93 for q in session.scalars(stmt).unique(): 

94 try: 

95 dupe_update = update_meta_row_stmt(q) 

96 res = conn.execute(dupe_update) 

97 m = "Execute update question meta statement for Q ID # %s : updated %s rows" 

98 log.info(m, q.id, res.rowcount) 

99 except Exception as exc: 

100 log.error( 

101 "Failed to update meta table for question Id %s with error: %s", 

102 q.id, 

103 exc, 

104 )