Coverage for netbox_data_import/api/views.py: 98%

1# SPDX-License-Identifier: Apache-2.0 

2# Copyright (C) 2025 Marcin Zieba <marcinpsk@gmail.com> 

3"""DRF viewsets for the data-import plugin API.""" 

4 

5from django.http import Http404 

6from netbox.api.viewsets import NetBoxModelViewSet 

7from rest_framework import viewsets, permissions 

8from rest_framework.permissions import DjangoModelPermissions 

9 

10from ..models import ( 

11 locked_profile_policy, 

12 locked_resolution_policy, 

13 ImportProfile, 

14 ColumnMapping, 

15 ClassRoleMapping, 

16 DeviceTypeMapping, 

17 IgnoredDevice, 

18 ColumnTransformRule, 

19 SourceResolution, 

20 ImportExecution, 

21) 

22from .serializers import ( 

23 ImportProfileSerializer, 

24 ColumnMappingSerializer, 

25 ClassRoleMappingSerializer, 

26 DeviceTypeMappingSerializer, 

27 IgnoredDeviceSerializer, 

28 ColumnTransformRuleSerializer, 

29 SourceResolutionSerializer, 

30 ImportExecutionSerializer, 

31) 

32 

33 

34class DjangoModelPermissionsWithView(DjangoModelPermissions): 

35 """Extends DjangoModelPermissions to require view_* permission for GET requests. 

36 

37 The stock DjangoModelPermissions does not map GET to any model permission, 

38 so list/retrieve endpoints are accessible to any authenticated user. This 

39 subclass closes that gap. 

40 """ 

41 

42 perms_map = { 

43 **DjangoModelPermissions.perms_map, 

44 "GET": ["%(app_label)s.view_%(model_name)s"], 

45 "HEAD": ["%(app_label)s.view_%(model_name)s"], 

46 "OPTIONS": [], 

47 } 

48 

49 

50class ImportProfileViewSet(NetBoxModelViewSet): 

51 """CRUD viewset for ImportProfile (NetBoxModel).""" 

52 

53 queryset = ImportProfile.objects.prefetch_related( 

54 "tags", "column_mappings", "class_role_mappings", "device_type_mappings" 

55 ) 

56 serializer_class = ImportProfileSerializer 

57 

58 

59class _PluginModelViewSet(viewsets.ModelViewSet): 

60 """Base class for plain-model viewsets in this plugin.""" 

61 

62 permission_classes = [permissions.IsAuthenticated, DjangoModelPermissionsWithView] 

63 

64 

65class ColumnMappingViewSet(_PluginModelViewSet): 

66 """CRUD viewset for ColumnMapping.""" 

67 

68 queryset = ColumnMapping.objects.select_related("profile") 

69 serializer_class = ColumnMappingSerializer 

70 

71 def get_queryset(self): 

72 """Filter by profile_id query param if provided.""" 

73 qs = super().get_queryset() 

74 profile_id = self.request.query_params.get("profile_id") 

75 if profile_id: 

76 qs = qs.filter(profile_id=profile_id) 

77 return qs 

78 

79 

80class ClassRoleMappingViewSet(_PluginModelViewSet): 

81 """CRUD viewset for ClassRoleMapping.""" 

82 

83 queryset = ClassRoleMapping.objects.select_related("profile", "rack_type") 

84 serializer_class = ClassRoleMappingSerializer 

85 

86 def get_queryset(self): 

87 """Filter by profile_id query param if provided.""" 

88 qs = super().get_queryset() 

89 profile_id = self.request.query_params.get("profile_id") 

90 if profile_id: 

91 qs = qs.filter(profile_id=profile_id) 

92 return qs 

93 

94 

95class DeviceTypeMappingViewSet(_PluginModelViewSet): 

96 """CRUD viewset for DeviceTypeMapping.""" 

97 

98 queryset = DeviceTypeMapping.objects.select_related("profile") 

99 serializer_class = DeviceTypeMappingSerializer 

100 

101 def get_queryset(self): 

102 """Filter by profile_id query param if provided.""" 

103 qs = super().get_queryset() 

104 profile_id = self.request.query_params.get("profile_id") 

105 if profile_id: 

106 qs = qs.filter(profile_id=profile_id) 

107 return qs 

108 

109 

110class IgnoredDeviceViewSet(_PluginModelViewSet): 

111 """CRUD viewset for IgnoredDevice.""" 

112 

113 queryset = IgnoredDevice.objects.select_related("profile") 

114 serializer_class = IgnoredDeviceSerializer 

115 

116 def get_queryset(self): 

117 """Filter by profile_id query param if provided.""" 

118 qs = super().get_queryset() 

119 profile_id = self.request.query_params.get("profile_id") 

120 if profile_id: 

121 qs = qs.filter(profile_id=profile_id) 

122 return qs 

123 

124 

125class ColumnTransformRuleViewSet(_PluginModelViewSet): 

126 """CRUD viewset for ColumnTransformRule.""" 

127 

128 queryset = ColumnTransformRule.objects.select_related("profile") 

129 serializer_class = ColumnTransformRuleSerializer 

130 

131 def get_queryset(self): 

132 """Filter by profile_id query param if provided.""" 

133 qs = super().get_queryset() 

134 profile_id = self.request.query_params.get("profile_id") 

135 if profile_id: 

136 qs = qs.filter(profile_id=profile_id) 

137 return qs 

138 

139 

140def _revalidate_against_the_stored_row(serializer): 

141 """Read the resolution again and check the request against the row as it now stands. 

142 

143 save() writes every field, so a request that changed another field first would otherwise be 

144 undone. The whole validation runs again rather than validate() alone, because the field checks 

145 read the stored row too, and the profile lock makes this reading of it authoritative. The 

146 result is discarded: the values are the request's own, which the first pass already holds. 

147 """ 

148 serializer.instance = SourceResolution.objects.get(pk=serializer.instance.pk) 

149 serializer.run_validation(serializer.initial_data) 

150 

151 

152class SourceResolutionViewSet(_PluginModelViewSet): 

153 """CRUD viewset for SourceResolution (rerere).""" 

154 

155 queryset = SourceResolution.objects.select_related("profile") 

156 serializer_class = SourceResolutionSerializer 

157 

158 # Each write serializes against an executing import, which holds the same profile row. 

159 def perform_create(self, serializer): 

160 """Create the resolution under the profile lock.""" 

161 try: 

162 with locked_profile_policy(serializer.validated_data["profile"].pk): 

163 serializer.save() 

164 except ImportProfile.DoesNotExist: 

165 # The profile is read to validate the request, and can be deleted before the lock. 

166 raise Http404 from None 

167 

168 def perform_update(self, serializer): 

169 """Update the resolution under its profile lock.""" 

170 # ValidatedModelSerializer.validate() writes the request values onto the instance, so only 

171 # its primary key still names the stored row. 

172 try: 

173 with locked_resolution_policy(serializer.instance.pk): 

174 _revalidate_against_the_stored_row(serializer) 

175 serializer.save() 

176 except (SourceResolution.DoesNotExist, ImportProfile.DoesNotExist): 

177 raise Http404 from None 

178 

179 def perform_destroy(self, instance): 

180 """Delete the resolution under its profile lock.""" 

181 try: 

182 with locked_resolution_policy(instance.pk): 

183 instance.delete() 

184 except (SourceResolution.DoesNotExist, ImportProfile.DoesNotExist): 

185 raise Http404 from None 

186 

187 def get_queryset(self): 

188 """Filter by profile_id query param if provided.""" 

189 qs = super().get_queryset() 

190 profile_id = self.request.query_params.get("profile_id") 

191 if profile_id: 

192 qs = qs.filter(profile_id=profile_id) 

193 return qs 

194 

195 

196class ImportExecutionViewSet(viewsets.ReadOnlyModelViewSet): 

197 """Read-only viewset for the Import Execution audit history.""" 

198 

199 queryset = ImportExecution.objects.select_related("profile") 

200 serializer_class = ImportExecutionSerializer 

201 permission_classes = [permissions.IsAuthenticated, DjangoModelPermissionsWithView] 

202 

203 def get_queryset(self): 

204 """Filter by profile_id query param if provided.""" 

205 qs = super().get_queryset() 

206 profile_id = self.request.query_params.get("profile_id") 

207 if profile_id: 

208 qs = qs.filter(profile_id=profile_id) 

209 return qs