Coverage for src/mesh/models/file_models.py: 86%
101 statements
« prev ^ index » next coverage.py v7.7.0, created at 2025-04-28 07:45 +0000
1from __future__ import annotations
3import os
4from collections.abc import Callable
5from typing import TYPE_CHECKING
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
13from mesh.model.file_helpers import (
14 FILE_DELETE_DEFAULT_ERROR,
15 FILE_NAME_MAX_LENGTH,
16 MeshFileSystemStorage,
17 file_name,
18)
20from .base_models import BaseChangeTrackingModel
22if TYPE_CHECKING: 22 ↛ 23line 22 didn't jump to line 23 because the condition on line 22 was never true
23 from mesh.model.roles.role_handler import RoleHandler
26MEGABYTE_VALUE = 1048576
29class BaseModelWithFiles(BaseChangeTrackingModel):
30 """
31 Model template for a model with attached files using the below BaseFileWrapperModel
32 model.
33 """
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 = []
41 class Meta:
42 abstract = True
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.
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
56def get_upload_path_from_model(instance: models.Model, filename: str) -> str:
57 """
58 Calls the `get_upload_path` method of the instance's model.
59 """
60 model_class = instance.__class__
61 upload_path_func = getattr(model_class, "get_upload_path", None)
62 if not upload_path_func or not callable(upload_path_func): 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true
63 raise Exception(
64 f"Error: model {model_class} does not have a callable `get_upload_path` method."
65 )
66 return upload_path_func(instance, filename)
69class BaseFileWrapperModel(BaseChangeTrackingModel):
70 """
71 Base model class providing a wrapper around a single file field.
72 This should be used instead of creating direct FileField on the models.
73 """
75 # Allowed file extensions
76 file_extensions = [".pdf", ".docx"]
77 # Max length for file name - Must be synced with the `name` field and
78 # the file name + the prefix path should be less than FILE_NAME_MAX_LENGTH
79 # 256 characters should be okay for any file.
80 file_name_max_length = 256
81 # File's max size (in bytes)
82 file_max_size = 10 * MEGABYTE_VALUE
84 file = models.FileField(
85 max_length=FILE_NAME_MAX_LENGTH,
86 upload_to=get_upload_path_from_model,
87 storage=MeshFileSystemStorage,
88 # validators=[file_validators], No validators - They are handled by forms
89 blank=False,
90 null=False,
91 )
92 # Stores the original file name. The file might be stored with a different name
93 # on the server. Note that you must update this field when changing the `file`
94 # field manually (it doesn't happen, usually a new Wrapper instance is created).
95 name = models.CharField(max_length=256, blank=True, null=False)
97 # The foreign key when the file is linked to another entity
98 # (Should always be the case)
99 attached_to: None | models.OneToOneField | models.ForeignKey = None
101 class Meta:
102 abstract = True
104 @classmethod
105 def get_help_text(cls) -> dict:
106 """ """
107 descriptions = []
108 if cls.file_extensions: 108 ↛ 110line 108 didn't jump to line 110 because the condition on line 108 was always true
109 descriptions.append(f"Allowed files are {', '.join(cls.file_extensions)}.")
110 descriptions.append(f"Maximum size: {cls.file_max_size / MEGABYTE_VALUE:.2f}Mb")
111 return {"type": "list", "texts": descriptions}
113 @staticmethod
114 def get_upload_path(instance: BaseFileWrapperModel, filename: str) -> str:
115 """
116 Customizable hook to get the file upload path from the model instance.
118 Note: We can't use @abstractmethod decorator here because we can't use
119 ABCMeta as the metaclass
120 """
121 raise NotImplementedError
123 @classmethod
124 def reverse_file_path(cls, file_path: str) -> BaseFileWrapperModel | None:
125 """
126 Returns the model instance associated to the given file path if the path
127 corresponds to a file of the current model.
129 Note: We can't use @abstractmethod decorator here because we can't use
130 ABCMeta as the metaclass
131 """
132 raise NotImplementedError
134 @staticmethod
135 def instance_valid_file(instance: BaseFileWrapperModel):
136 return instance and instance.file and getattr(instance.file, "url", False)
138 @classmethod
139 def run_file_validators(cls, value: UploadedFile):
140 for validator in cls.file_validators():
141 validator(value)
143 @classmethod
144 def file_validators(cls) -> list[Callable]:
145 """
146 Returns all the model's file validators.
147 Default is the 3 base validators: file_extension, file_name_length and
148 file_size.
149 """
150 return [
151 cls.validate_file_extension,
152 cls.validate_file_name_length,
153 cls.validate_file_size,
154 ]
156 @classmethod
157 def validate_file_extension(cls, value: UploadedFile):
158 """
159 Validate the file extension wrt. the class requirements.
160 All file types are allowed if `file_extensions` is empty.
161 """
162 if not cls.file_extensions:
163 return
164 extension = os.path.splitext(value.name)[1]
165 if extension.lower() not in cls.file_extensions:
166 raise ValidationError(
167 f"Unsupported file extension: {extension.lower()}. "
168 f"Valid file extension(s) are: {', '.join(cls.file_extensions)}"
169 )
171 @classmethod
172 def validate_file_name_length(cls, value: UploadedFile):
173 name = file_name(value.name)
174 max_length = min(cls.file_name_max_length, FILE_NAME_MAX_LENGTH)
175 if len(name) > max_length:
176 raise ValidationError(
177 f"The provided file name is too long: {name}. "
178 f"The maximum allowed length is {max_length} characters."
179 )
181 @classmethod
182 def validate_file_size(cls, value: UploadedFile):
183 if value.size > cls.file_max_size:
184 raise ValidationError(
185 f"The provided file is too large: {value.size / (MEGABYTE_VALUE):.2f} MB. "
186 f"Maximum allowed file size is {cls.file_max_size / (MEGABYTE_VALUE):.2f} MB."
187 )
189 def save(self, *args, **kwargs) -> None:
190 """
191 Deletes the old file from the filesystem if the database insertion/update
192 succeeds.
194 TODO: This is not robust with rollbacks. In case this is called during
195 an encapsulating transaction that gets rolled back, this might create
196 an inconsistent status : the file is effectively deleted but the database entry
197 still points to the deleted file as it was rolled back...
198 """
199 # Set the file name now if it does not exist - It might be changed later by the storage
200 self.name = self.name or file_name(self.file.name)
202 original_path = None
203 # Check if there's already a instance of this model with the current pk.
204 # Store the old file path for potential deletion.
205 if self.pk:
206 try:
207 original_model = self.__class__.objects.get(pk=self.pk)
208 original_path = original_model.file.path
209 except self.__class__.DoesNotExist:
210 pass
212 super().save(*args, **kwargs)
214 # If the path changed, delete the old file.
215 if (
216 original_path is not None
217 and self.file.path != original_path
218 and os.path.exists(original_path)
219 ):
220 os.remove(original_path)
222 def delete(self, *args, **kwargs) -> tuple[int, dict[str, int]]:
223 """
224 Deletes the file from the filesystem if the database deletion succeeds.
225 """
226 file_to_delete = self.file
228 result = super().delete(*args, **kwargs)
230 if file_to_delete: 230 ↛ 233line 230 didn't jump to line 233 because the condition on line 230 was always true
231 file_to_delete.delete(save=False)
233 return result
235 def check_access_right(self, role_handler: RoleHandler, right_code: str) -> bool:
236 """
237 Customizable hook to check the given access right.
238 Only used when serving a file.
239 Default is False.
240 """
241 return False
243 def get_file_url(self) -> str:
244 """
245 Returns the URL to the model's file.
246 """
247 if not self.file.url:
248 return ""
249 file_identifier = encode_for_url(self.file.name)
250 return reverse_lazy(
251 "mesh:serve_protected_file", kwargs={"file_identifier": file_identifier}
252 )