Coverage for src / mesh / models / orm / file_models.py: 97%
87 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-05-04 12:41 +0000
1from __future__ import annotations
3import os
4from collections.abc import Callable
6from django.core.exceptions import ValidationError
7from django.core.files.uploadedfile import UploadedFile
8from django.db import models
9from ptf.url_utils import encode_for_url
11from mesh.models.file_helpers import (
12 FILE_DELETE_DEFAULT_ERROR,
13 FILE_NAME_MAX_LENGTH,
14 MeshFileSystemStorage,
15 file_name,
16)
18from .base_models import BaseChangeTrackingModel
20MEGABYTE_VALUE = 1048576
23class BaseModelWithFiles(BaseChangeTrackingModel):
24 """
25 Model template for a model with attached files using the below BaseFileWrapperModel
26 model.
27 """
29 # List of the required file fields. We need to mention them somewhere because
30 # some "required" file fields can have null=True for other purpose
31 file_fields_required = []
32 # List of file fields or related file models that can be deleted by an user.
33 file_fields_deletable = []
35 class Meta: # type: ignore
36 abstract = True
38 def can_delete_file(self, file_field: str) -> tuple[bool, str]:
39 """
40 Check function that returns whether the provided file field is deletable.
41 Basic implementation only checks if the field is in the deletable fields list.
43 Returns:
44 - result: bool Whether the field can be deleted
45 - error_message: str An explaining error message when `result=False`
46 """
47 return file_field in self.file_fields_deletable, FILE_DELETE_DEFAULT_ERROR
50def get_upload_path_from_model(instance: BaseFileWrapperModel, filename: str) -> str:
51 """
52 Calls the `get_upload_path` method of the instance's model.
53 """
54 return instance.get_upload_path(filename)
57class BaseFileWrapperModel(BaseChangeTrackingModel):
58 """
59 Base model class providing a wrapper around a single file field.
60 This should be used instead of creating direct FileField on the models.
61 """
63 class Meta:
64 abstract = True
66 def __class_getitem__(cls, item):
67 return cls
69 # Allowed file extensions
70 file_extensions = [".pdf", ".docx"]
71 # Max length for file name - Must be synced with the `name` field and
72 # the file name + the prefix path should be less than FILE_NAME_MAX_LENGTH
73 # 256 characters should be okay for any file.
74 file_name_max_length = 256
75 # File's max size (in bytes)
76 file_max_size = 10 * MEGABYTE_VALUE
78 file = models.FileField(
79 max_length=FILE_NAME_MAX_LENGTH,
80 upload_to=get_upload_path_from_model,
81 storage=MeshFileSystemStorage,
82 # validators=[file_validators], No validators - They are handled by forms
83 blank=False,
84 null=False,
85 )
86 # Stores the original file name. The file might be stored with a different name
87 # on the server. Note that you must update this field when changing the `file`
88 # field manually (it doesn't happen, usually a new Wrapper instance is created).
89 name = models.CharField(max_length=256, null=False)
91 # The foreign key when the file is linked to another entity
92 # (Should always be the case)
93 # attached_to: models.ForeignKey[_T]
95 @classmethod
96 def get_help_text(cls) -> dict:
97 """ """
98 descriptions = []
99 if cls.file_extensions:
100 descriptions.append(f"Allowed files are {', '.join(cls.file_extensions)}.")
101 descriptions.append(f"Maximum size: {cls.file_max_size / MEGABYTE_VALUE:.2f}Mb")
102 return {"type": "list", "texts": descriptions}
104 def get_upload_path(self, filename: str) -> str:
105 """
106 Customizable hook to get the file upload path from the model instance.
108 Note: We can't use @abstractmethod decorator here because we can't use
109 ABCMeta as the metaclass
110 """
111 raise NotImplementedError
113 @staticmethod
114 def instance_valid_file(instance: BaseFileWrapperModel):
115 return instance and instance.file and getattr(instance.file, "url", False)
117 @classmethod
118 def run_file_validators(cls, value: UploadedFile):
119 for validator in cls.file_validators():
120 validator(value)
122 @classmethod
123 def file_validators(cls) -> list[Callable]:
124 """
125 Returns all the model's file validators.
126 Default is the 3 base validators: file_extension, file_name_length and
127 file_size.
128 """
129 return [
130 cls.validate_file_extension,
131 cls.validate_file_name_length,
132 cls.validate_file_size,
133 ]
135 @classmethod
136 def validate_file_extension(cls, value: UploadedFile):
137 """
138 Validate the file extension wrt. the class requirements.
139 All file types are allowed if `file_extensions` is empty.
140 """
141 if not cls.file_extensions:
142 return
143 extension = os.path.splitext(value.name)[1]
144 if extension.lower() not in cls.file_extensions:
145 raise ValidationError(
146 f"Unsupported file extension: {extension.lower()}. "
147 f"Valid file extension(s) are: {', '.join(cls.file_extensions)}"
148 )
150 @classmethod
151 def validate_file_name_length(cls, value: UploadedFile):
152 name = file_name(value.name)
153 max_length = min(cls.file_name_max_length, FILE_NAME_MAX_LENGTH)
154 if len(name) > max_length:
155 raise ValidationError(
156 f"The provided file name is too long: {name}. "
157 f"The maximum allowed length is {max_length} characters."
158 )
160 @classmethod
161 def validate_file_size(cls, value: UploadedFile):
162 if value.size > cls.file_max_size:
163 raise ValidationError(
164 f"The provided file is too large: {value.size / (MEGABYTE_VALUE):.2f} MB. "
165 f"Maximum allowed file size is {cls.file_max_size / (MEGABYTE_VALUE):.2f} MB."
166 )
168 def save(self, *args, **kwargs) -> None:
169 """
170 Deletes the old file from the filesystem if the database insertion/update
171 succeeds.
173 TODO: This is not robust with rollbacks. In case this is called during
174 an encapsulating transaction that gets rolled back, this might create
175 an inconsistent status : the file is effectively deleted but the database entry
176 still points to the deleted file as it was rolled back...
177 """
178 # Set the file name now if it does not exist - It might be changed later by the storage
179 self.name = self.name or file_name(self.file.name)
181 original_path = None
182 # Check if there's already a instance of this model with the current pk.
183 # Store the old file path for potential deletion.
184 if self.pk:
185 try:
186 original_model = self.__class__.objects.get(pk=self.pk)
187 original_path = original_model.file.path
188 except self.__class__.DoesNotExist:
189 pass
191 super().save(*args, **kwargs)
193 # If the path changed, delete the old file.
194 if (
195 original_path is not None
196 and self.file.path != original_path
197 and os.path.exists(original_path)
198 ):
199 os.remove(original_path)
201 def delete(self, *args, **kwargs) -> tuple[int, dict[str, int]]:
202 """
203 Deletes the file from the filesystem if the database deletion succeeds.
204 """
205 file_to_delete = self.file
207 result = super().delete(*args, **kwargs)
209 if file_to_delete:
210 file_to_delete.delete(save=False)
212 return result
214 def check_access_right(self, role, right_code: str) -> bool:
215 """
216 Customizable hook to check the given access right.
217 Only used when serving a file.
218 Default is False.
219 """
220 return False
222 def get_file_identifier(self):
223 return encode_for_url(self.file.name)