Coverage for src / mesh / models / orm / file_models.py: 97%
87 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-10-22 07:20 +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 descriptions = []
98 if cls.file_extensions:
99 descriptions.append(f"Allowed files are {', '.join(cls.file_extensions)}.")
100 descriptions.append(f"Maximum size: {cls.file_max_size / MEGABYTE_VALUE:.2f}Mb")
101 return {"type": "list", "texts": descriptions}
103 def get_upload_path(self, filename: str) -> str:
104 """
105 Customizable hook to get the file upload path from the model instance.
107 Note: We can't use @abstractmethod decorator here because we can't use
108 ABCMeta as the metaclass
109 """
110 raise NotImplementedError
112 @staticmethod
113 def instance_valid_file(instance: BaseFileWrapperModel):
114 return instance and instance.file and getattr(instance.file, "url", False)
116 @classmethod
117 def run_file_validators(cls, value: UploadedFile):
118 for validator in cls.file_validators():
119 validator(value)
121 @classmethod
122 def file_validators(cls) -> list[Callable]:
123 """
124 Returns all the model's file validators.
125 Default is the 3 base validators: file_extension, file_name_length and
126 file_size.
127 """
128 return [
129 cls.validate_file_extension,
130 cls.validate_file_name_length,
131 cls.validate_file_size,
132 ]
134 @classmethod
135 def validate_file_extension(cls, value: UploadedFile):
136 """
137 Validate the file extension wrt. the class requirements.
138 All file types are allowed if `file_extensions` is empty.
139 """
140 if not cls.file_extensions:
141 return
142 extension = os.path.splitext(value.name)[1]
143 if extension.lower() not in cls.file_extensions:
144 raise ValidationError(
145 f"Unsupported file extension: {extension.lower()}. "
146 f"Valid file extension(s) are: {', '.join(cls.file_extensions)}"
147 )
149 @classmethod
150 def validate_file_name_length(cls, value: UploadedFile):
151 name = file_name(value.name)
152 max_length = min(cls.file_name_max_length, FILE_NAME_MAX_LENGTH)
153 if len(name) > max_length:
154 raise ValidationError(
155 f"The provided file name is too long: {name}. "
156 f"The maximum allowed length is {max_length} characters."
157 )
159 @classmethod
160 def validate_file_size(cls, value: UploadedFile):
161 if value.size > cls.file_max_size:
162 raise ValidationError(
163 f"The provided file is too large: {value.size / (MEGABYTE_VALUE):.2f} MB. "
164 f"Maximum allowed file size is {cls.file_max_size / (MEGABYTE_VALUE):.2f} MB."
165 )
167 def save(self, *args, **kwargs) -> None:
168 """
169 Deletes the old file from the filesystem if the database insertion/update
170 succeeds.
172 TODO: This is not robust with rollbacks. In case this is called during
173 an encapsulating transaction that gets rolled back, this might create
174 an inconsistent status : the file is effectively deleted but the database entry
175 still points to the deleted file as it was rolled back...
176 """
177 # Set the file name now if it does not exist - It might be changed later by the storage
178 self.name = self.name or file_name(self.file.name)
180 original_path = None
181 # Check if there's already a instance of this model with the current pk.
182 # Store the old file path for potential deletion.
183 if self.pk:
184 try:
185 original_model = self.__class__.objects.get(pk=self.pk)
186 original_path = original_model.file.path
187 except self.__class__.DoesNotExist:
188 pass
190 super().save(*args, **kwargs)
192 # If the path changed, delete the old file.
193 if (
194 original_path is not None
195 and self.file.path != original_path
196 and os.path.exists(original_path)
197 ):
198 os.remove(original_path)
200 def delete(self, *args, **kwargs) -> tuple[int, dict[str, int]]:
201 """
202 Deletes the file from the filesystem if the database deletion succeeds.
203 """
204 file_to_delete = self.file
206 result = super().delete(*args, **kwargs)
208 if file_to_delete:
209 file_to_delete.delete(save=False)
211 return result
213 def check_access_right(self, role, right_code: str) -> bool:
214 """
215 Customizable hook to check the given access right.
216 Only used when serving a file.
217 Default is False.
218 """
219 return False
221 def get_file_identifier(self):
222 return encode_for_url(self.file.name)