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

1import logging 

2from hashlib import md5 

3from typing import List, Optional 

4 

5from sqlalchemy import types, ForeignKey, Index, select 

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

7from sqlalchemy.dialects.mysql import insert 

8 

9from .meta import Base 

10from rfpy.model.questionnaire import QuestionDefinition, QuestionInstance 

11 

12 

13log = logging.getLogger(__name__) 

14 

15 

16class QuestionMeta(Base): 

17 __tablename__ = "questions_meta" 

18 __table_args__ = ( 

19 Index("ft_alltext", "alltext", mysql_prefix="FULLTEXT"), 

20 {"mysql_engine": "InnoDB", "mysql_charset": "utf8mb4"}, 

21 ) 

22 

23 question_id: Mapped[int] = mapped_column( 

24 types.Integer, 

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

26 unique=True, 

27 nullable=False, # Ensure question_id cannot be NULL 

28 ) 

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

30 "QuestionDefinition", 

31 back_populates="meta", 

32 viewonly=True, 

33 ) 

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

35 "QuestionInstance", 

36 secondary="questions", 

37 back_populates="meta", 

38 viewonly=True, 

39 ) 

40 

41 # For efficient & simplified full text querying of questions 

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

43 

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

45 # similar to the yesno reports developed for T2R 

46 # Support for 'questions like this' functionality 

47 signature: Mapped[Optional[str]] = mapped_column(types.VARCHAR(length=64), index=True, nullable=True) 

48 

49 def __repr__(self) -> str: 

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

51 

52 

53def qsignature(sig_list: List[str]) -> str: 

54 return md5("|".join(sig_list).encode("utf8")).hexdigest() 

55 

56 

57def update_meta_row_stmt(q): 

58 qtext_set = [q.title] 

59 sig_list = [] 

60 for el in q.elements: 

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

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

63 qtext_set.append(el.label) 

64 if el.choices: 

65 for c in el.choices: 

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

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

68 qtext = " ".join(qtext_set) 

69 sig = qsignature(sig_list) 

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

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

72 

73 

74def update_qmeta_table(session: Session, qdef_ids): 

75 """ 

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

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

78 

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

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

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

82 """ 

83 conn = session.connection() 

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

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

86 try: 

87 dupe_update = update_meta_row_stmt(q) 

88 res = conn.execute(dupe_update) 

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

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

91 except Exception as exc: 

92 log.error( 

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

94 q.id, 

95 exc, 

96 )