Coverage for  / home / runner / work / netbox-data-import-plugin / netbox-data-import-plugin / netbox-data-import-plugin / netbox_data_import / api / views.py: 100%

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 netbox.api.viewsets import NetBoxModelViewSet 

6from rest_framework import viewsets, permissions 

7 

8from ..models import ( 

9 ImportProfile, 

10 ColumnMapping, 

11 ClassRoleMapping, 

12 DeviceTypeMapping, 

13 IgnoredDevice, 

14 ColumnTransformRule, 

15 SourceResolution, 

16 ImportJob, 

17) 

18from .serializers import ( 

19 ImportProfileSerializer, 

20 ColumnMappingSerializer, 

21 ClassRoleMappingSerializer, 

22 DeviceTypeMappingSerializer, 

23 IgnoredDeviceSerializer, 

24 ColumnTransformRuleSerializer, 

25 SourceResolutionSerializer, 

26 ImportJobSerializer, 

27) 

28 

29 

30class ImportProfileViewSet(NetBoxModelViewSet): 

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

32 

33 queryset = ImportProfile.objects.prefetch_related( 

34 "tags", 

35 "column_mappings", 

36 "class_role_mappings", 

37 "device_type_mappings", 

38 ) 

39 serializer_class = ImportProfileSerializer 

40 

41 

42class _PluginModelViewSet(viewsets.ModelViewSet): 

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

44 

45 permission_classes = [permissions.IsAuthenticated] 

46 

47 

48class ColumnMappingViewSet(_PluginModelViewSet): 

49 """CRUD viewset for ColumnMapping.""" 

50 

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

52 serializer_class = ColumnMappingSerializer 

53 

54 def get_queryset(self): 

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

56 qs = super().get_queryset() 

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

58 if profile_id: 

59 qs = qs.filter(profile_id=profile_id) 

60 return qs 

61 

62 

63class ClassRoleMappingViewSet(_PluginModelViewSet): 

64 """CRUD viewset for ClassRoleMapping.""" 

65 

66 queryset = ClassRoleMapping.objects.select_related("profile") 

67 serializer_class = ClassRoleMappingSerializer 

68 

69 def get_queryset(self): 

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

71 qs = super().get_queryset() 

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

73 if profile_id: 

74 qs = qs.filter(profile_id=profile_id) 

75 return qs 

76 

77 

78class DeviceTypeMappingViewSet(_PluginModelViewSet): 

79 """CRUD viewset for DeviceTypeMapping.""" 

80 

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

82 serializer_class = DeviceTypeMappingSerializer 

83 

84 def get_queryset(self): 

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

86 qs = super().get_queryset() 

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

88 if profile_id: 

89 qs = qs.filter(profile_id=profile_id) 

90 return qs 

91 

92 

93class IgnoredDeviceViewSet(_PluginModelViewSet): 

94 """CRUD viewset for IgnoredDevice.""" 

95 

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

97 serializer_class = IgnoredDeviceSerializer 

98 

99 def get_queryset(self): 

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

101 qs = super().get_queryset() 

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

103 if profile_id: 

104 qs = qs.filter(profile_id=profile_id) 

105 return qs 

106 

107 

108class ColumnTransformRuleViewSet(_PluginModelViewSet): 

109 """CRUD viewset for ColumnTransformRule.""" 

110 

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

112 serializer_class = ColumnTransformRuleSerializer 

113 

114 def get_queryset(self): 

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

116 qs = super().get_queryset() 

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

118 if profile_id: 

119 qs = qs.filter(profile_id=profile_id) 

120 return qs 

121 

122 

123class SourceResolutionViewSet(_PluginModelViewSet): 

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

125 

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

127 serializer_class = SourceResolutionSerializer 

128 

129 def get_queryset(self): 

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

131 qs = super().get_queryset() 

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

133 if profile_id: 

134 qs = qs.filter(profile_id=profile_id) 

135 return qs 

136 

137 

138class ImportJobViewSet(viewsets.ReadOnlyModelViewSet): 

139 """Read-only viewset for ImportJob history.""" 

140 

141 queryset = ImportJob.objects.select_related("profile") 

142 serializer_class = ImportJobSerializer 

143 permission_classes = [permissions.IsAuthenticated] 

144 

145 def get_queryset(self): 

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

147 qs = super().get_queryset() 

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

149 if profile_id: 

150 qs = qs.filter(profile_id=profile_id) 

151 return qs