Coverage for src / mesh / models / file_models.py: 90%

94 statements  

« prev     ^ index     » next       coverage.py v7.13.1, created at 2026-02-20 10:03 +0000

1from __future__ import annotations 

2 

3import os 

4from collections.abc import Callable 

5from typing import TYPE_CHECKING, Generic, TypeVar 

6 

7from django.core.exceptions import ValidationError 

8from django.core.files.uploadedfile import UploadedFile 

9from django.db import models 

10from django.urls import reverse_lazy 

11from ptf.url_utils import encode_for_url 

12 

13from mesh.model.file_helpers import ( 

14 FILE_DELETE_DEFAULT_ERROR, 

15 FILE_NAME_MAX_LENGTH, 

16 MeshFileSystemStorage, 

17 file_name, 

18) 

19 

20from .base_models import BaseChangeTrackingModel 

21 

22if TYPE_CHECKING: 

23 from mesh.model.roles.role_handler import RoleHandler 

24 

25 

26MEGABYTE_VALUE = 1048576 

27 

28 

29class BaseModelWithFiles(BaseChangeTrackingModel): 

30 """ 

31 Model template for a model with attached files using the below BaseFileWrapperModel 

32 model. 

33 """ 

34 

35 # List of the required file fields. We need to mention them somewhere because 

36 # some "required" file fields can have null=True for other purpose 

37 file_fields_required = [] 

38 # List of file fields or related file models that can be deleted by an user. 

39 file_fields_deletable = [] 

40 

41 class Meta: # type: ignore 

42 abstract = True 

43 

44 def can_delete_file(self, file_field: str) -> tuple[bool, str]: 

45 """ 

46 Check function that returns whether the provided file field is deletable. 

47 Basic implementation only checks if the field is in the deletable fields list. 

48 

49 Returns: 

50 - result: bool Whether the field can be deleted 

51 - error_message: str An explaining error message when `result=False` 

52 """ 

53 return file_field in self.file_fields_deletable, FILE_DELETE_DEFAULT_ERROR 

54 

55 

56def get_upload_path_from_model(instance: BaseFileWrapperModel, filename: str) -> str: 

57 """ 

58 Calls the `get_upload_path` method of the instance's model. 

59 """ 

60 return instance.get_upload_path(filename) 

61 

62 

63_T = TypeVar("_T", bound=BaseModelWithFiles) 

64 

65 

66class BaseFileWrapperModel(BaseChangeTrackingModel, Generic[_T]): 

67 """ 

68 Base model class providing a wrapper around a single file field. 

69 This should be used instead of creating direct FileField on the models. 

70 """ 

71 

72 # Allowed file extensions 

73 file_extensions = [".pdf", ".docx"] 

74 # Max length for file name - Must be synced with the `name` field and 

75 # the file name + the prefix path should be less than FILE_NAME_MAX_LENGTH 

76 # 256 characters should be okay for any file. 

77 file_name_max_length = 256 

78 # File's max size (in bytes) 

79 file_max_size = 10 * MEGABYTE_VALUE 

80 

81 file = models.FileField( 

82 max_length=FILE_NAME_MAX_LENGTH, 

83 upload_to=get_upload_path_from_model, 

84 storage=MeshFileSystemStorage, 

85 # validators=[file_validators], No validators - They are handled by forms 

86 blank=False, 

87 null=False, 

88 ) 

89 # Stores the original file name. The file might be stored with a different name 

90 # on the server. Note that you must update this field when changing the `file` 

91 # field manually (it doesn't happen, usually a new Wrapper instance is created). 

92 name = models.CharField(max_length=256, null=False) 

93 

94 # The foreign key when the file is linked to another entity 

95 # (Should always be the case) 

96 # attached_to: models.ForeignKey[_T] 

97 

98 class Meta: # type: ignore 

99 abstract = True 

100 

101 @classmethod 

102 def get_help_text(cls) -> dict: 

103 """ """ 

104 descriptions = [] 

105 if cls.file_extensions: 

106 descriptions.append(f"Allowed files are {', '.join(cls.file_extensions)}.") 

107 descriptions.append(f"Maximum size: {cls.file_max_size / MEGABYTE_VALUE:.2f}Mb") 

108 return {"type": "list", "texts": descriptions} 

109 

110 def get_upload_path(self, filename: str) -> str: 

111 """ 

112 Customizable hook to get the file upload path from the model instance. 

113 

114 Note: We can't use @abstractmethod decorator here because we can't use 

115 ABCMeta as the metaclass 

116 """ 

117 raise NotImplementedError 

118 

119 @classmethod 

120 def reverse_file_path(cls, file_path: str) -> BaseFileWrapperModel | None: 

121 """ 

122 Returns the model instance associated to the given file path if the path 

123 corresponds to a file of the current model. 

124 

125 Note: We can't use @abstractmethod decorator here because we can't use 

126 ABCMeta as the metaclass 

127 """ 

128 raise NotImplementedError 

129 

130 @staticmethod 

131 def instance_valid_file(instance: BaseFileWrapperModel): 

132 return instance and instance.file and getattr(instance.file, "url", False) 

133 

134 @classmethod 

135 def run_file_validators(cls, value: UploadedFile): 

136 for validator in cls.file_validators(): 

137 validator(value) 

138 

139 @classmethod 

140 def file_validators(cls) -> list[Callable]: 

141 """ 

142 Returns all the model's file validators. 

143 Default is the 3 base validators: file_extension, file_name_length and 

144 file_size. 

145 """ 

146 return [ 

147 cls.validate_file_extension, 

148 cls.validate_file_name_length, 

149 cls.validate_file_size, 

150 ] 

151 

152 @classmethod 

153 def validate_file_extension(cls, value: UploadedFile): 

154 """ 

155 Validate the file extension wrt. the class requirements. 

156 All file types are allowed if `file_extensions` is empty. 

157 """ 

158 if not cls.file_extensions: 

159 return 

160 extension = os.path.splitext(value.name)[1] 

161 if extension.lower() not in cls.file_extensions: 

162 raise ValidationError( 

163 f"Unsupported file extension: {extension.lower()}. " 

164 f"Valid file extension(s) are: {', '.join(cls.file_extensions)}" 

165 ) 

166 

167 @classmethod 

168 def validate_file_name_length(cls, value: UploadedFile): 

169 name = file_name(value.name) 

170 max_length = min(cls.file_name_max_length, FILE_NAME_MAX_LENGTH) 

171 if len(name) > max_length: 

172 raise ValidationError( 

173 f"The provided file name is too long: {name}. " 

174 f"The maximum allowed length is {max_length} characters." 

175 ) 

176 

177 @classmethod 

178 def validate_file_size(cls, value: UploadedFile): 

179 if value.size > cls.file_max_size: 

180 raise ValidationError( 

181 f"The provided file is too large: {value.size / (MEGABYTE_VALUE):.2f} MB. " 

182 f"Maximum allowed file size is {cls.file_max_size / (MEGABYTE_VALUE):.2f} MB." 

183 ) 

184 

185 def save(self, *args, **kwargs) -> None: 

186 """ 

187 Deletes the old file from the filesystem if the database insertion/update 

188 succeeds. 

189 

190 TODO: This is not robust with rollbacks. In case this is called during 

191 an encapsulating transaction that gets rolled back, this might create 

192 an inconsistent status : the file is effectively deleted but the database entry 

193 still points to the deleted file as it was rolled back... 

194 """ 

195 # Set the file name now if it does not exist - It might be changed later by the storage 

196 self.name = self.name or file_name(self.file.name) 

197 

198 original_path = None 

199 # Check if there's already a instance of this model with the current pk. 

200 # Store the old file path for potential deletion. 

201 if self.pk: 

202 try: 

203 original_model = self.__class__.objects.get(pk=self.pk) 

204 original_path = original_model.file.path 

205 except self.__class__.DoesNotExist: 

206 pass 

207 

208 super().save(*args, **kwargs) 

209 

210 # If the path changed, delete the old file. 

211 if ( 

212 original_path is not None 

213 and self.file.path != original_path 

214 and os.path.exists(original_path) 

215 ): 

216 os.remove(original_path) 

217 

218 def delete(self, *args, **kwargs) -> tuple[int, dict[str, int]]: 

219 """ 

220 Deletes the file from the filesystem if the database deletion succeeds. 

221 """ 

222 file_to_delete = self.file 

223 

224 result = super().delete(*args, **kwargs) 

225 

226 if file_to_delete: 

227 file_to_delete.delete(save=False) 

228 

229 return result 

230 

231 def check_access_right(self, role_handler: RoleHandler, right_code: str) -> bool: 

232 """ 

233 Customizable hook to check the given access right. 

234 Only used when serving a file. 

235 Default is False. 

236 """ 

237 return False 

238 

239 def get_file_url(self) -> str: 

240 """ 

241 Returns the URL to the model's file. 

242 """ 

243 if not self.file.url: 

244 return "" 

245 file_identifier = encode_for_url(self.file.name) 

246 return reverse_lazy( 

247 "mesh:serve_protected_file", kwargs={"file_identifier": file_identifier} 

248 )