Coverage for core/models/files.py: 61%

1import logging 

2import os 

3from functools import cached_property 

4 

5from django.conf import settings 

6from django.core.exceptions import ValidationError 

7from django.core.files.storage import storages 

8from django.db import models 

9from django.utils.translation import gettext as _ 

10 

11from extras.storage import ScriptFileSystemStorage 

12from netbox.models.features import SyncedDataMixin 

13from utilities.querysets import RestrictedQuerySet 

14 

15from ..choices import ManagedFileRootPathChoices 

16 

17__all__ = ( 

18 'ManagedFile', 

19) 

20 

21logger = logging.getLogger('netbox.core.files') 

22 

23 

24class ManagedFile(SyncedDataMixin, models.Model): 

25 """ 

26 Database representation for a file on disk. This class is typically wrapped by a proxy class (e.g. ScriptModule) 

27 to provide additional functionality. 

28 """ 

29 created = models.DateTimeField( 

30 verbose_name=_('created'), 

31 auto_now_add=True 

32 ) 

33 last_updated = models.DateTimeField( 

34 verbose_name=_('last updated'), 

35 editable=False, 

36 blank=True, 

37 null=True 

38 ) 

39 file_root = models.CharField( 

40 verbose_name=_('file root'), 

41 max_length=1000, 

42 choices=ManagedFileRootPathChoices 

43 ) 

44 file_path = models.FilePathField( 

45 verbose_name=_('file path'), 

46 editable=False, 

47 help_text=_('File path relative to the designated root path') 

48 ) 

49 

50 objects = RestrictedQuerySet.as_manager() 

51 _netbox_private = True 

52 

53 class Meta: 

54 ordering = ('file_root', 'file_path') 

55 constraints = ( 

56 models.UniqueConstraint( 

57 fields=('file_root', 'file_path'), 

58 name='%(app_label)s_%(class)s_unique_root_path' 

59 ), 

60 ) 

61 verbose_name = _('managed file') 

62 verbose_name_plural = _('managed files') 

63 

64 def __str__(self): 

65 return self.name 

66 

67 @property 

68 def name(self): 

69 return self.file_path 

70 

71 @property 

72 def full_path(self): 

73 return os.path.join(self._resolve_root_path(), self.file_path) 

74 

75 def _resolve_root_path(self): 

76 storage = self.storage 

77 if isinstance(storage, ScriptFileSystemStorage): 

78 return { 

79 'scripts': settings.SCRIPTS_ROOT, 

80 'reports': settings.REPORTS_ROOT, 

81 }[self.file_root] 

82 return "" 

83 

84 def sync_data(self): 

85 if self.data_file: 

86 self.file_path = os.path.basename(self.data_path) 

87 

88 storage = self.storage 

89 

90 with storage.open(self.full_path, 'wb+') as new_file: 

91 new_file.write(self.data_file.data) 

92 sync_data.alters_data = True 

93 

94 @cached_property 

95 def storage(self): 

96 return storages.create_storage(storages.backends["scripts"]) 

97 

98 def clean(self): 

99 super().clean() 

100 

101 if self.data_file and not self.file_path: 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true

102 self.file_path = os.path.basename(self.data_path) 

103 

104 # Ensure that the file root and path make a unique pair 

105 if self._meta.model.objects.filter( 105 ↛ 108line 105 didn't jump to line 108 because the condition on line 105 was never true

106 file_root=self.file_root, file_path=self.file_path 

107 ).exclude(pk=self.pk).exists(): 

108 raise ValidationError( 

109 _("A {model} with this file path already exists ({path}).").format( 

110 model=self._meta.verbose_name.lower(), 

111 path=f"{self.file_root}/{self.file_path}" 

112 )) 

113 

114 def delete(self, *args, **kwargs): 

115 # Delete file from disk 

116 storage = self.storage 

117 try: 

118 storage.delete(self.full_path) 

119 except FileNotFoundError: 

120 pass 

121 

122 return super().delete(*args, **kwargs)