mirror of
https://github.com/inventree/InvenTree.git
synced 2026-05-22 01:06:50 +00:00
[feature] Rename attachments (#11920)
* Implementation * Update API and CHANGELOG * Annotate response type * Simplify attachment renaming - Use the existing API endpoint * Capture the actual saved path * Tweak attachment table fields * Use built-in validation * Update docs * Unit testing * Ignore some lines from coverage * Check if file exists before deleting
This commit is contained in:
@@ -1,11 +1,14 @@
|
||||
"""InvenTree API version information."""
|
||||
|
||||
# InvenTree API version
|
||||
INVENTREE_API_VERSION = 487
|
||||
INVENTREE_API_VERSION = 488
|
||||
"""Increment this API version number whenever there is a significant change to the API that any clients need to know about."""
|
||||
|
||||
INVENTREE_API_TEXT = """
|
||||
|
||||
v488 -> 2026-05-17 : https://github.com/inventree/InvenTree/pull/11920
|
||||
- Allow renaming of attachments after upload via the API
|
||||
|
||||
v487 -> 2026-05-15 : https://github.com/inventree/InvenTree/pull/11948
|
||||
- Make SelectionList default nullable
|
||||
- Add icon to TreePath schema
|
||||
|
||||
@@ -661,7 +661,9 @@ class InvenTreeAttachmentMixin(InvenTreePermissionCheckMixin):
|
||||
|
||||
Before deleting the model instance, delete any associated attachments.
|
||||
"""
|
||||
self.attachments.all().delete()
|
||||
for attachment in list(self.attachments.all()):
|
||||
attachment.delete()
|
||||
|
||||
super().delete(*args, **kwargs)
|
||||
|
||||
@property
|
||||
|
||||
@@ -781,8 +781,34 @@ class AttachmentList(AttachmentMixin, BulkDeleteMixin, ListCreateAPI):
|
||||
class AttachmentDetail(AttachmentMixin, RetrieveUpdateDestroyAPI):
|
||||
"""Detail API endpoint for Attachment objects."""
|
||||
|
||||
def update(self, request, *args, **kwargs):
|
||||
"""Update an existing attachment object."""
|
||||
attachment = self.get_object()
|
||||
|
||||
if not attachment.check_permission('change', request.user):
|
||||
raise PermissionDenied(
|
||||
_('User does not have permission to edit this attachment')
|
||||
)
|
||||
|
||||
partial = kwargs.pop('partial', False)
|
||||
data = self.clean_data(request.data)
|
||||
|
||||
# Extract filename first
|
||||
filename = data.pop('filename', None)
|
||||
|
||||
# Run other validation / updates first, before attempting to rename the file
|
||||
serializer = self.get_serializer(attachment, data=data, partial=partial)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
self.perform_update(serializer)
|
||||
|
||||
# User is attempting to rename the file
|
||||
if filename and attachment.basename and filename != attachment.basename:
|
||||
attachment.rename(filename)
|
||||
|
||||
return Response(serializer.data)
|
||||
|
||||
def destroy(self, request, *args, **kwargs):
|
||||
"""Check user permissions before deleting an attachment."""
|
||||
"""Delete an existing attachment object."""
|
||||
attachment = self.get_object()
|
||||
|
||||
if not attachment.check_permission('delete', request.user):
|
||||
|
||||
@@ -15,6 +15,7 @@ from datetime import timedelta, timezone
|
||||
from email.utils import make_msgid
|
||||
from enum import Enum
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
from secrets import compare_digest
|
||||
from typing import Any, Optional
|
||||
|
||||
@@ -25,9 +26,10 @@ from django.contrib.contenttypes.fields import GenericForeignKey
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from django.contrib.humanize.templatetags.humanize import naturaltime
|
||||
from django.core.cache import cache
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.core.exceptions import SuspiciousFileOperation, ValidationError
|
||||
from django.core.files.base import ContentFile
|
||||
from django.core.files.storage import default_storage
|
||||
from django.core.files.utils import validate_file_name
|
||||
from django.core.mail import EmailMultiAlternatives, get_connection
|
||||
from django.core.mail.utils import DNS_NAME
|
||||
from django.core.validators import MinLengthValidator, MinValueValidator
|
||||
@@ -1948,6 +1950,22 @@ class Attachment(InvenTree.models.MetadataMixin, InvenTree.models.InvenTreeModel
|
||||
|
||||
choice_fnc = common.validators.attachment_model_options
|
||||
|
||||
def delete(self, *args, **kwargs):
|
||||
"""Custom delete method for the Attachment model.
|
||||
|
||||
- Ensure that the attached file is deleted from storage when the database entry is removed
|
||||
"""
|
||||
attachment = self.attachment
|
||||
|
||||
super().delete(*args, **kwargs)
|
||||
|
||||
if attachment and default_storage.exists(attachment.name):
|
||||
try:
|
||||
# Remove the attached file from storage
|
||||
default_storage.delete(attachment.name)
|
||||
except Exception: # pragma: no cover
|
||||
pass
|
||||
|
||||
def save(self, *args, **kwargs):
|
||||
"""Custom 'save' method for the Attachment model.
|
||||
|
||||
@@ -1993,6 +2011,60 @@ class Attachment(InvenTree.models.MetadataMixin, InvenTree.models.InvenTreeModel
|
||||
return os.path.basename(self.attachment.name)
|
||||
return str(self.link)
|
||||
|
||||
def validate_rename(self, filename: str):
|
||||
"""Validate that the provided filename is valid, for renaming an attachment."""
|
||||
filename = filename.strip()
|
||||
|
||||
if not self.attachment:
|
||||
raise ValidationError(_('No file attached to rename'))
|
||||
|
||||
if not filename:
|
||||
raise ValidationError(_('Filename cannot be empty'))
|
||||
|
||||
try:
|
||||
validate_file_name(filename, allow_relative_path=False)
|
||||
except SuspiciousFileOperation:
|
||||
raise ValidationError(_('Invalid filename'))
|
||||
|
||||
current_ext = os.path.splitext(self.attachment.name)[1]
|
||||
new_ext = os.path.splitext(filename)[1]
|
||||
|
||||
if current_ext.lower() != new_ext.lower():
|
||||
raise ValidationError(_('Cannot change file extension'))
|
||||
|
||||
def rename(self, filename: str):
|
||||
"""Rename the attached file."""
|
||||
self.validate_rename(filename)
|
||||
|
||||
old_path = Path(self.attachment.name)
|
||||
new_path = old_path.parent / filename
|
||||
|
||||
if old_path == new_path: # pragma: no cover
|
||||
# No change in filename
|
||||
return
|
||||
|
||||
if not new_path.is_relative_to(old_path.parent): # pragma: no cover
|
||||
raise ValidationError(_('Invalid filename'))
|
||||
|
||||
new_path = new_path.as_posix()
|
||||
|
||||
if default_storage.exists(new_path):
|
||||
raise ValidationError(_('A file with this name already exists'))
|
||||
|
||||
# Create a new file with the new name, and delete the old file
|
||||
new_path = default_storage.save(new_path, self.attachment.file)
|
||||
|
||||
# Ensure that the new file exists
|
||||
if not default_storage.exists(new_path): # pragma: no cover
|
||||
raise ValidationError(_('Failed to save renamed file'))
|
||||
|
||||
# Update the database file path
|
||||
self.attachment.name = new_path
|
||||
self.save()
|
||||
|
||||
# Remove the old path
|
||||
default_storage.delete(old_path)
|
||||
|
||||
model_type = models.CharField(
|
||||
max_length=100,
|
||||
validators=[common.validators.validate_attachment_model_type],
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""JSON serializers for common components."""
|
||||
"""API serializers for common components."""
|
||||
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from django.db.models import Count, OuterRef, Subquery
|
||||
|
||||
@@ -1,5 +1,9 @@
|
||||
"""API unit tests for InvenTree common functionality."""
|
||||
|
||||
import io
|
||||
|
||||
from django.core.files.base import ContentFile
|
||||
from django.core.files.storage import default_storage
|
||||
from django.urls import reverse
|
||||
|
||||
import common.models
|
||||
@@ -675,3 +679,113 @@ class ParameterAPITests(InvenTreeAPITestCase):
|
||||
self.assertFalse(
|
||||
common.models.Parameter.objects.filter(template=template.pk).exists()
|
||||
)
|
||||
|
||||
|
||||
class AttachmentAPITests(InvenTreeAPITestCase):
|
||||
"""Tests for the Attachment API."""
|
||||
|
||||
def test_attachments(self):
|
||||
"""Test API functionality for attachments."""
|
||||
from common.models import Attachment
|
||||
from part.models import Part
|
||||
|
||||
self.assignRole('part.add')
|
||||
|
||||
part = Part.objects.create(name='Test Part', description='A part for testing')
|
||||
|
||||
N = Attachment.objects.count()
|
||||
|
||||
# Upload multiple attachments against the part instance
|
||||
for ii in range(5):
|
||||
file_object = io.StringIO('Hello world')
|
||||
file_object.seek(0)
|
||||
|
||||
fn = f'test_file_{ii}.txt'
|
||||
|
||||
content_file = ContentFile(file_object.read(), name=fn)
|
||||
|
||||
url = reverse('api-attachment-list')
|
||||
|
||||
response = self.post(
|
||||
url,
|
||||
data={
|
||||
'model_type': 'part',
|
||||
'model_id': part.pk,
|
||||
'attachment': content_file,
|
||||
'comment': f'This is test file {ii}',
|
||||
},
|
||||
format='multipart',
|
||||
expected_code=201,
|
||||
)
|
||||
|
||||
data = response.data
|
||||
|
||||
# Check that the file has actually been created
|
||||
self.assertEqual(data['filename'], fn)
|
||||
self.assertTrue(
|
||||
default_storage.exists(data['attachment'].replace('/media/', ''))
|
||||
)
|
||||
|
||||
# Check that we have the expected number of attachments
|
||||
self.assertEqual(Attachment.objects.count(), N + 5)
|
||||
self.assertEqual(part.attachments.count(), 5)
|
||||
|
||||
# Let's rename one of the attachments
|
||||
att = part.attachments.first()
|
||||
self.assertEqual(att.basename, 'test_file_0.txt')
|
||||
|
||||
url = reverse('api-attachment-detail', kwargs={'pk': att.pk})
|
||||
|
||||
# A few failed attempts
|
||||
for new_name in [
|
||||
'different_ext.docx',
|
||||
'test_file_1.txt',
|
||||
'../../test_file.txt',
|
||||
]:
|
||||
print('- ATTEMPTING:', new_name)
|
||||
response = self.patch(url, data={'filename': new_name}, expected_code=400)
|
||||
|
||||
att.refresh_from_db()
|
||||
self.assertEqual(att.basename, 'test_file_0.txt')
|
||||
|
||||
# Let's try seriously this time
|
||||
new_name = 'a_new_file.txt'
|
||||
response = self.patch(url, data={'filename': new_name}, expected_code=200)
|
||||
|
||||
att.refresh_from_db()
|
||||
self.assertEqual(att.basename, new_name)
|
||||
|
||||
# Check that the file has been renamed on disk
|
||||
self.assertTrue(
|
||||
default_storage.exists(f'attachments/part/{part.pk}/{new_name}')
|
||||
)
|
||||
self.assertFalse(
|
||||
default_storage.exists(f'attachments/part/{part.pk}/test_file_0.txt')
|
||||
)
|
||||
|
||||
# Next, let's delete the attachment manually - via the API
|
||||
response = self.delete(url, expected_code=403)
|
||||
self.assignRole('part.delete')
|
||||
response = self.delete(url, expected_code=204)
|
||||
|
||||
# Check that the file has been deleted from disk
|
||||
self.assertFalse(
|
||||
default_storage.exists(f'attachments/part/{part.pk}/{new_name}')
|
||||
)
|
||||
|
||||
self.assertEqual(Attachment.objects.count(), N + 4)
|
||||
self.assertEqual(part.attachments.count(), 4)
|
||||
|
||||
# Fetch the remaining attachments
|
||||
attachments = list(part.attachments.all())
|
||||
|
||||
# Now, delete the part instance
|
||||
part.active = False
|
||||
part.save()
|
||||
part.delete()
|
||||
|
||||
self.assertEqual(Attachment.objects.count(), N)
|
||||
|
||||
for att in attachments:
|
||||
# Ensure that the file associated with each attachment has been removed
|
||||
self.assertFalse(default_storage.exists(att.attachment.path))
|
||||
|
||||
Reference in New Issue
Block a user