Coverage for  / home / runner / work / netbox-data-import-plugin / netbox-data-import-plugin / netbox-data-import-plugin / netbox_data_import / models.py: 99%

1# SPDX-License-Identifier: Apache-2.0 

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

3from django.db import models 

4from django.urls import reverse 

5from netbox.models import NetBoxModel 

6 

7PREVIEW_VIEW_CHOICES = [ 

8 ("rows", "Row view"), 

9 ("racks", "Rack view"), 

10] 

11 

12 

13TARGET_FIELD_CHOICES = [ 

14 ("rack_name", "Rack name"), 

15 ("device_name", "Device name"), 

16 ("device_class", "Device class (maps to role/rack)"), 

17 ("face", "Face (Front/Back)"), 

18 ("airflow", "Airflow"), 

19 ("u_position", "U position"), 

20 ("status", "Status"), 

21 ("make", "Make (manufacturer)"), 

22 ("model", "Model (device type)"), 

23 ("u_height", "U height"), 

24 ("serial", "Serial number"), 

25 ("asset_tag", "Asset tag"), 

26 ("source_id", "Source ID (stored in custom field)"), 

27] 

28 

29 

30class ImportProfile(NetBoxModel): 

31 """Named configuration for one source file format.""" 

32 

33 name = models.CharField(max_length=100, unique=True) 

34 description = models.TextField(blank=True) 

35 sheet_name = models.CharField( 

36 max_length=100, 

37 default="Data", 

38 help_text="Name of the Excel worksheet to read", 

39 ) 

40 source_id_column = models.CharField( 

41 max_length=100, 

42 blank=True, 

43 help_text="Column whose value is stored in a NetBox custom field (e.g. 'Id')", 

44 ) 

45 custom_field_name = models.CharField( 

46 max_length=100, 

47 blank=True, 

48 help_text="NetBox custom field name to store the source ID in (e.g. 'cans_id')", 

49 ) 

50 update_existing = models.BooleanField( 

51 default=True, 

52 help_text="Update existing NetBox objects when a match is found", 

53 ) 

54 create_missing_device_types = models.BooleanField( 

55 default=True, 

56 help_text="Auto-create manufacturers and device types that don't exist in NetBox", 

57 ) 

58 preview_view_mode = models.CharField( 

59 max_length=10, 

60 choices=PREVIEW_VIEW_CHOICES, 

61 default="rows", 

62 help_text="How to display the import preview (row table or rack diagrams)", 

63 ) 

64 

65 # Override tags reverse accessor to avoid clashes with other plugins 

66 tags = models.ManyToManyField( 

67 to="extras.Tag", 

68 related_name="+", 

69 blank=True, 

70 ) 

71 

72 class Meta: 

73 ordering = ["name"] 

74 verbose_name = "Import Profile" 

75 verbose_name_plural = "Import Profiles" 

76 

77 def __str__(self): 

78 return self.name 

79 

80 def get_absolute_url(self): 

81 """Return the detail URL for this import profile.""" 

82 return reverse("plugins:netbox_data_import:importprofile", args=[self.pk]) 

83 

84 

85class ColumnMapping(models.Model): 

86 """Maps one source column header to one semantic NetBox field.""" 

87 

88 profile = models.ForeignKey( 

89 ImportProfile, 

90 on_delete=models.CASCADE, 

91 related_name="column_mappings", 

92 ) 

93 source_column = models.CharField( 

94 max_length=200, 

95 help_text="Exact column header in the source file (case-sensitive)", 

96 ) 

97 target_field = models.CharField(max_length=50, choices=TARGET_FIELD_CHOICES) 

98 

99 class Meta: 

100 ordering = ["profile", "target_field"] 

101 unique_together = [("profile", "target_field")] 

102 verbose_name = "Column Mapping" 

103 verbose_name_plural = "Column Mappings" 

104 

105 def __str__(self): 

106 return f"{self.source_column} → {self.get_target_field_display()}" 

107 

108 def get_absolute_url(self): 

109 """Return the edit URL for this column mapping.""" 

110 return reverse("plugins:netbox_data_import:columnmapping_edit", args=[self.pk]) 

111 

112 

113class ClassRoleMapping(models.Model): 

114 """Maps a source 'class' value to a NetBox outcome (rack or device role).""" 

115 

116 profile = models.ForeignKey( 

117 ImportProfile, 

118 on_delete=models.CASCADE, 

119 related_name="class_role_mappings", 

120 ) 

121 source_class = models.CharField( 

122 max_length=200, 

123 help_text="Value from the class column (e.g. 'Server', 'Cabinet')", 

124 ) 

125 creates_rack = models.BooleanField( 

126 default=False, 

127 help_text="If checked, rows with this class create a Rack instead of a Device", 

128 ) 

129 role_slug = models.CharField( 

130 max_length=100, 

131 blank=True, 

132 help_text="NetBox device role slug (ignored when 'creates rack' is checked)", 

133 ) 

134 ignore = models.BooleanField( 

135 default=False, 

136 help_text="If checked, rows with this class are silently skipped (not shown as errors)", 

137 ) 

138 

139 class Meta: 

140 ordering = ["profile", "source_class"] 

141 unique_together = [("profile", "source_class")] 

142 verbose_name = "Class → Role Mapping" 

143 verbose_name_plural = "Class → Role Mappings" 

144 

145 def __str__(self): 

146 if self.creates_rack: 

147 return f"{self.source_class} → Rack" 

148 return f"{self.source_class} → {self.role_slug}" 

149 

150 def get_absolute_url(self): 

151 """Return the edit URL for this class→role mapping.""" 

152 return reverse("plugins:netbox_data_import:classrolemapping_edit", args=[self.pk]) 

153 

154 

155class ImportJob(models.Model): 

156 """Records a completed import run with its results.""" 

157 

158 profile = models.ForeignKey( 

159 ImportProfile, 

160 on_delete=models.SET_NULL, 

161 null=True, 

162 blank=True, 

163 related_name="import_jobs", 

164 ) 

165 created = models.DateTimeField(auto_now_add=True) 

166 input_filename = models.CharField(max_length=255, blank=True) 

167 dry_run = models.BooleanField(default=False) 

168 site_name = models.CharField(max_length=100, blank=True) 

169 result_counts = models.JSONField(default=dict) 

170 result_rows = models.JSONField(default=list) 

171 

172 class Meta: 

173 ordering = ["-created"] 

174 verbose_name = "Import Job" 

175 verbose_name_plural = "Import Jobs" 

176 

177 def __str__(self): 

178 return f"Import {self.pk} — {self.created:%Y-%m-%d %H:%M} ({self.input_filename})" 

179 

180 def get_absolute_url(self): 

181 """Return the associated profile's URL (no per-job detail view exists).""" 

182 if not self.profile_id: 

183 return reverse("plugins:netbox_data_import:importprofile_list") 

184 return reverse("plugins:netbox_data_import:importprofile", args=[self.profile_id]) 

185 

186 

187class DeviceTypeMapping(models.Model): 

188 """Explicit (make, model) override when source naming doesn't slugify cleanly.""" 

189 

190 profile = models.ForeignKey( 

191 ImportProfile, 

192 on_delete=models.CASCADE, 

193 related_name="device_type_mappings", 

194 ) 

195 source_make = models.CharField(max_length=200) 

196 source_model = models.CharField(max_length=200) 

197 netbox_manufacturer_slug = models.CharField(max_length=100) 

198 netbox_device_type_slug = models.CharField(max_length=100) 

199 

200 class Meta: 

201 ordering = ["profile", "source_make", "source_model"] 

202 unique_together = [("profile", "source_make", "source_model")] 

203 verbose_name = "Device Type Mapping" 

204 verbose_name_plural = "Device Type Mappings" 

205 

206 def __str__(self): 

207 return ( 

208 f"{self.source_make} / {self.source_model} → {self.netbox_manufacturer_slug}/{self.netbox_device_type_slug}" 

209 ) 

210 

211 def get_absolute_url(self): 

212 """Return the edit URL for this device type mapping.""" 

213 return reverse("plugins:netbox_data_import:devicetypemapping_edit", args=[self.pk]) 

214 

215 

216class ManufacturerMapping(models.Model): 

217 """Maps a source 'make' value to an existing NetBox manufacturer slug.""" 

218 

219 profile = models.ForeignKey( 

220 ImportProfile, 

221 on_delete=models.CASCADE, 

222 related_name="manufacturer_mappings", 

223 ) 

224 source_make = models.CharField( 

225 max_length=200, 

226 help_text="Exact source make value (e.g. 'Dell EMC')", 

227 ) 

228 netbox_manufacturer_slug = models.CharField( 

229 max_length=100, 

230 help_text="NetBox manufacturer slug to map this make to (e.g. 'dell')", 

231 ) 

232 

233 class Meta: 

234 ordering = ["profile", "source_make"] 

235 unique_together = [("profile", "source_make")] 

236 verbose_name = "Manufacturer Mapping" 

237 verbose_name_plural = "Manufacturer Mappings" 

238 

239 def __str__(self): 

240 return f"{self.source_make} → {self.netbox_manufacturer_slug}" 

241 

242 

243class IgnoredDevice(models.Model): 

244 """Per-device ignore record — prevents a specific source device from being imported.""" 

245 

246 profile = models.ForeignKey( 

247 ImportProfile, 

248 on_delete=models.CASCADE, 

249 related_name="ignored_devices", 

250 ) 

251 source_id = models.CharField( 

252 max_length=200, 

253 help_text="Source ID value that identifies this device", 

254 ) 

255 device_name = models.CharField( 

256 max_length=200, 

257 blank=True, 

258 help_text="Original device name (for display only)", 

259 ) 

260 

261 class Meta: 

262 ordering = ["profile", "source_id"] 

263 unique_together = [("profile", "source_id")] 

264 verbose_name = "Ignored Device" 

265 verbose_name_plural = "Ignored Devices" 

266 

267 def __str__(self): 

268 return f"{self.device_name or self.source_id} (ignored)" 

269 

270 

271class ColumnTransformRule(models.Model): 

272 r"""Regex-based transform applied to a source column during parse. 

273 

274 Example: source_column='Name', pattern='^(\w{4,8}) - (.+)$', 

275 group_1_target='asset_tag', group_2_target='device_name' 

276 transforms "59AH76 - PROD-LAB03-SW1" into asset_tag="59AH76", device_name="PROD-LAB03-SW1". 

277 """ 

278 

279 profile = models.ForeignKey( 

280 ImportProfile, 

281 on_delete=models.CASCADE, 

282 related_name="column_transform_rules", 

283 ) 

284 source_column = models.CharField( 

285 max_length=200, 

286 help_text="Source Excel column to transform (exact header name)", 

287 ) 

288 pattern = models.CharField( 

289 max_length=500, 

290 help_text=r"Python regex with capture groups (re.fullmatch). E.g. ^(\w+) - (.+)$", 

291 ) 

292 group_1_target = models.CharField( 

293 max_length=50, 

294 blank=True, 

295 choices=TARGET_FIELD_CHOICES, 

296 help_text="Target field for capture group 1 (leave blank to ignore)", 

297 ) 

298 group_2_target = models.CharField( 

299 max_length=50, 

300 blank=True, 

301 choices=TARGET_FIELD_CHOICES, 

302 help_text="Target field for capture group 2 (leave blank to ignore)", 

303 ) 

304 

305 class Meta: 

306 ordering = ["profile", "source_column"] 

307 unique_together = [("profile", "source_column")] 

308 verbose_name = "Column Transform Rule" 

309 verbose_name_plural = "Column Transform Rules" 

310 

311 def __str__(self): 

312 return f"{self.source_column}: {self.pattern}" 

313 

314 def get_absolute_url(self): 

315 """Return the edit URL for this column transform rule.""" 

316 return reverse("plugins:netbox_data_import:columntransformrule_edit", args=[self.pk]) 

317 

318 

319class SourceResolution(models.Model): 

320 """Saved manual resolution for a specific source cell value. 

321 

322 When a user manually splits "59AH76 - PROD-LAB03-SW1" into asset_tag=59AH76 

323 and device_name=PROD-LAB03-SW1, that resolution is saved here. On re-import, 

324 parse_file applies it automatically (like git rerere). 

325 """ 

326 

327 profile = models.ForeignKey( 

328 ImportProfile, 

329 on_delete=models.CASCADE, 

330 related_name="source_resolutions", 

331 ) 

332 source_id = models.CharField( 

333 max_length=200, 

334 help_text="Source ID of the row this resolution applies to", 

335 ) 

336 source_column = models.CharField( 

337 max_length=200, 

338 help_text="Column name this resolution applies to", 

339 ) 

340 original_value = models.TextField( 

341 help_text="Original cell value before resolution", 

342 ) 

343 resolved_fields = models.JSONField( 

344 default=dict, 

345 help_text="Dict of target_field -> resolved_value (e.g. {'device_name': 'SW1', 'asset_tag': '59AH76'})", 

346 ) 

347 

348 class Meta: 

349 ordering = ["profile", "source_id"] 

350 unique_together = [("profile", "source_id", "source_column")] 

351 verbose_name = "Source Resolution" 

352 verbose_name_plural = "Source Resolutions" 

353 

354 def __str__(self): 

355 return f"{self.source_id}/{self.source_column}: {self.original_value!r}" 

356 

357 

358class DeviceExistingMatch(models.Model): 

359 """Explicit match between a source row and an existing NetBox device. 

360 

361 When a user clicks "Link existing" on a device preview row, this record is saved. 

362 On re-import, the engine uses this to emit action='update' against the matched device 

363 instead of action='create', even if the device has no source-ID custom field yet. 

364 """ 

365 

366 profile = models.ForeignKey( 

367 ImportProfile, 

368 on_delete=models.CASCADE, 

369 related_name="device_matches", 

370 ) 

371 source_id = models.CharField( 

372 max_length=200, 

373 help_text="Source ID value that identifies this row", 

374 ) 

375 source_asset_tag = models.CharField( 

376 max_length=100, 

377 blank=True, 

378 default="", 

379 help_text="Asset tag from source row (for display / lookup; may become stale)", 

380 ) 

381 netbox_device_id = models.PositiveIntegerField( 

382 help_text="Primary key of the matched NetBox Device", 

383 ) 

384 device_name = models.CharField( 

385 max_length=200, 

386 blank=True, 

387 help_text="NetBox device name (for display only; may become stale)", 

388 ) 

389 

390 class Meta: 

391 ordering = ["profile", "source_id"] 

392 unique_together = [("profile", "source_id")] 

393 verbose_name = "Device Existing Match" 

394 verbose_name_plural = "Device Existing Matches" 

395 

396 def __str__(self): 

397 tag = f" / {self.source_asset_tag}" if self.source_asset_tag else "" 

398 return f"{self.source_id}{tag} → Device #{self.netbox_device_id} ({self.device_name})"