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

1import logging 

2from hashlib import md5 

3from typing import List 

4 

5from sqlalchemy import types, Column, ForeignKey, Index 

6from sqlalchemy.orm import backref, relationship 

7from sqlalchemy.dialects.mysql import insert 

8 

9from .meta import Base 

10from .questionnaire import QuestionDefinition 

11 

12log = logging.getLogger(__name__) 

13 

14 

15class QuestionMeta(Base): 

16 __tablename__ = "questions_meta" 

17 __table_args__ = ( 

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

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

20 ) 

21 

22 question_id = Column( 

23 types.Integer, 

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

25 unique=True, 

26 nullable=False 

27 ) 

28 question_def = relationship( 

29 'QuestionDefinition', 

30 backref=backref( 

31 'meta', 

32 uselist=False, 

33 cascade='all,delete', 

34 passive_deletes=True 

35 ) 

36 ) 

37 question_instance = relationship( 

38 'QuestionInstance', 

39 secondary=QuestionDefinition.__table__, 

40 sync_backref=False, 

41 backref=backref('meta', uselist=False, viewonly=True), 

42 viewonly=True 

43 ) 

44 

45 # For efficient & simplified full text querying of questions 

46 alltext = Column(types.TEXT) 

47 

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

49 # similar to the yesno reports developed for T2R 

50 # Support for 'questions like this' functionality 

51 signature = Column(types.VARCHAR(64), index=True) 

52 

53 def __repr__(self) -> str: 

54 return f'<QuestionMeta, QDef # {self.question_id}, signature: {self.signature}>' 

55 

56 

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

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

59 

60 

61def update_meta_row_stmt(q): 

62 qtext_set = [q.title] 

63 sig_list = [] 

64 for el in q.elements: 

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

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

67 qtext_set.append(el.label) 

68 if el.choices: 

69 for c in el.choices: 

70 if 'label' in c and c['label'] is not None: 

71 qtext_set.append(c['label']) 

72 qtext = " ".join(qtext_set) 

73 sig = qsignature(sig_list) 

74 stmt = insert(QuestionMeta).values( 

75 question_id=q.id, 

76 alltext=qtext, 

77 signature=sig 

78 ) 

79 return stmt.on_duplicate_key_update( 

80 alltext=qtext, 

81 signature=sig 

82 ) 

83 

84 

85def update_qmeta_table(session, qdef_ids): 

86 ''' 

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

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

89 

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

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

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

93 ''' 

94 conn = session.connection() 

95 for q in (session.query(QuestionDefinition) 

96 .filter(QuestionDefinition.id.in_(qdef_ids))): 

97 try: 

98 dupe_update = update_meta_row_stmt(q) 

99 res = conn.execute(dupe_update) 

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

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

102 except Exception as exc: 

103 log.error('Failed to update meta table for question Id %s with error: %s', q.id, exc)