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

1from unicodedata import normalize 

2from typing import Sequence, Any 

3import mimetypes 

4import pathlib 

5import re 

6 

7from sqlalchemy import Integer, Unicode, event 

8from sqlalchemy.orm import ( 

9 object_session, 

10 validates, 

11 DeclarativeBase, 

12 Mapped, 

13 mapped_column, 

14) 

15 

16 

17def human_friendly_bytes(size): 

18 if not size or size == 0: 

19 return "0 KB" 

20 elif size < 1024: 

21 return "1 KB" 

22 elif size < 1024 * 1024: 

23 return "%s KB" % int(size / 1024) 

24 else: 

25 return "%s MB" % int(size / 1024 / 1024) 

26 

27 

28class Visitor: # pragma: no cover 

29 """ 

30 Base visitor classes enabling subclasses to implement 

31 just the methods they need. 

32 """ 

33 

34 def __init__(self): 

35 self.result = None 

36 

37 def hello_section(self, sec): 

38 pass 

39 

40 def goodbye_section(self, sec): 

41 pass 

42 

43 def visit_question(self, question): 

44 """ 

45 Using NotImplemented because presence of this method is 

46 used to determine whether or not to load questions 

47 

48 Therefore better to make it clear that this method defined in this 

49 class is never called 

50 """ 

51 raise NotImplementedError() 

52 

53 def get_result(self): 

54 return self.result 

55 

56 def finalise(self): 

57 pass 

58 

59 

60class Base(DeclarativeBase): 

61 """ 

62 @DynamicAttrs 

63 """ 

64 

65 # by default only show id 

66 public_attrs: Sequence = ["id"] 

67 

68 __table_args__: Any = ({"mysql_engine": "InnoDB", "mysql_charset": "utf8mb4"},) 

69 

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

71 

72 def as_dict(self): 

73 attrs = self.public_attrs 

74 return {k: getattr(self, k, "Not Provided") for k in attrs} 

75 

76 def __repr__(self): 

77 return f"<{self.__class__.__name__} {self.id}>" 

78 

79 @property 

80 def _instance_session(self): 

81 return object_session(self) 

82 

83 

84@event.listens_for(Base, "instrument_class", propagate=True) 

85def receive_before_mapper_configured(mapper, class_): 

86 """ 

87 Check that all classes use InnoDB and utf8mb4 charset. If not Foreign Key 

88 relationships can fail, especially during schema migrations 

89 """ 

90 dialect_options = class_.__table__.dialect_options 

91 if "mysql" not in dialect_options: 

92 raise ValueError("SQLAlchemy config error, no MySQL options for %s" % class_) 

93 mysql_options = dialect_options["mysql"] 

94 if mysql_options["engine"] != "InnoDB": 

95 raise ValueError( 

96 "SQLAlchemy config error, MySQL engine not InnoDB for %s" % class_ 

97 ) 

98 if mysql_options["charset"] != "utf8mb4": 

99 raise ValueError( 

100 "SQLAlchemy config error, MySQL charset not utf8mb4 for %s" % class_ 

101 ) 

102 

103 

104class AttachmentMixin: 

105 size_bytes: Mapped[int] = mapped_column("size", Integer, default=0, nullable=False) 

106 filename: Mapped[str] = mapped_column(Unicode(255), nullable=False) 

107 mimetype: Mapped[str] = mapped_column(Unicode(100), nullable=False) 

108 

109 @validates("filename") 

110 def _make_safe_filename(self, _attr_name, filename): 

111 """ 

112 Uploaded files can use trick filenames like '../../../rc.local' 

113 to try to hack an operating system. 

114 Filenames aren't used directly in this app, so shouldn't be a danger 

115 but we clean up anyway. 

116 """ 

117 ascii_filename = ( 

118 normalize("NFKD", filename).encode("ascii", "ignore").decode("ascii") 

119 ) 

120 ascii_filename = pathlib.Path(ascii_filename).name 

121 return re.sub(r"\s|/|\\", "_", ascii_filename) 

122 

123 def guess_set_mimetype(self, filename): 

124 """ 

125 Set the mimetype attribute for this file based on 

126 the filename extension 

127 """ 

128 mtype, _enc = mimetypes.guess_type(filename) 

129 self.mimetype = mtype 

130 return mtype 

131 

132 @property 

133 def size(self): 

134 return human_friendly_bytes(self.size_bytes) 

135 

136 def __repr__(self) -> str: 

137 return ( 

138 f"{self.__class__.__name__} - filename: {self.filename} size: {self.size}" 

139 )