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

1from unicodedata import normalize 

2import mimetypes 

3import pathlib 

4import re 

5 

6from sqlalchemy import ( 

7 Column, 

8 Integer, 

9 Unicode, 

10 event 

11) 

12from sqlalchemy.ext.declarative import declarative_base 

13from sqlalchemy.orm import object_session, validates 

14 

15 

16def human_friendly_bytes(size): 

17 if not size or size == 0: 

18 return '0 KB' 

19 elif size < 1024: 

20 return '1 KB' 

21 elif size < 1024 * 1024: 

22 return '%s KB' % int(size / 1024) 

23 else: 

24 return '%s MB' % int(size / 1024 / 1024) 

25 

26 

27class Visitor: # pragma: no cover 

28 

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 RfpBase(object): 

61 ''' 

62 @DynamicAttrs 

63 ''' 

64 # by default only show id 

65 public_attrs = ['id'] 

66 

67 __table_args__ = ( 

68 {'mysql_engine': 'InnoDB', 'mysql_charset': 'utf8mb4'}, 

69 ) 

70 

71 id = Column(Integer, primary_key=True) 

72 

73 def as_dict(self): 

74 attrs = self.public_attrs 

75 return {k: getattr(self, k, 'Not Provided') for k in attrs} 

76 

77 def __repr__(self): 

78 return f'<{self.__class__.__name__} {self.id}>' 

79 

80 @property 

81 def _instance_session(self): 

82 return object_session(self) 

83 

84 

85Base = declarative_base(cls=RfpBase) 

86 

87 

88@event.listens_for(Base, 'instrument_class', propagate=True) 

89def receive_before_mapper_configured(mapper, class_): 

90 ''' 

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

92 relationships can fail, especially during schema migrations 

93 ''' 

94 dialect_options = class_.__table__.dialect_options 

95 if 'mysql' not in dialect_options: 

96 raise ValueError('SQLAlchemy config error, no MySQL options for %s' % class_) 

97 mysql_options = dialect_options['mysql'] 

98 if mysql_options['engine'] != 'InnoDB': 

99 raise ValueError('SQLAlchemy config error, MySQL engine not InnoDB for %s' % class_) 

100 if mysql_options['charset'] != 'utf8mb4': 

101 raise ValueError('SQLAlchemy config error, MySQL charset not utf8mb4 for %s' % class_) 

102 

103 

104class AttachmentMixin(): 

105 

106 size_bytes = Column('size', Integer, default=0, nullable=False) 

107 filename = Column(Unicode(255), nullable=False) 

108 mimetype = Column(Unicode(100), nullable=False) 

109 

110 @validates('filename') 

111 def _make_safe_filename(self, _attr_name, filename): 

112 ''' 

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

114 to try to hack an operating system. 

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

116 but we clean up anyway. 

117 ''' 

118 ascii_filename = (normalize('NFKD', filename) 

119 .encode('ascii', 'ignore') 

120 .decode('ascii')) 

121 ascii_filename = pathlib.Path(ascii_filename).name 

122 return re.sub(r'\s|/|\\', '_', ascii_filename) 

123 

124 def guess_set_mimetype(self, filename): 

125 ''' 

126 Set the mimetype attribute for this file based on 

127 the filename extension 

128 ''' 

129 mtype, _enc = mimetypes.guess_type(filename) 

130 self.mimetype = mtype 

131 return mtype 

132 

133 @property 

134 def size(self): 

135 return human_friendly_bytes(self.size_bytes) 

136 

137 def __repr__(self) -> str: 

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