Coverage for core/views.py: 48%

1import json 

2import platform 

3from copy import deepcopy 

4 

5from django import __version__ as django_version 

6from django.apps import apps as django_apps_registry 

7from django.conf import settings 

8from django.contrib import messages 

9from django.contrib.auth.mixins import UserPassesTestMixin 

10from django.core.cache import cache 

11from django.db import DatabaseError, connection 

12from django.http import Http404, HttpResponse, HttpResponseForbidden 

13from django.shortcuts import get_object_or_404, redirect, render 

14from django.urls import reverse 

15from django.utils.decorators import method_decorator 

16from django.utils.http import content_disposition_header 

17from django.utils.translation import gettext_lazy as _ 

18from django.views.decorators.cache import never_cache 

19from django.views.generic import View 

20from django_rq.queues import get_queue_by_index, get_redis_connection 

21from django_rq.settings import get_queues_list, get_queues_map 

22from django_rq.utils import get_statistics 

23from rq.exceptions import NoSuchJobError 

24from rq.job import Job as RQ_Job 

25from rq.job import JobStatus as RQJobStatus 

26from rq.worker import Worker 

27from rq.worker_registration import clean_worker_registry 

28 

29from core.utils import ( 

30 delete_rq_job, 

31 enqueue_rq_job, 

32 get_db_schema, 

33 get_rq_jobs_from_status, 

34 requeue_rq_job, 

35 stop_rq_job, 

36) 

37from extras.ui.panels import CustomFieldsPanel, TagsPanel 

38from netbox.config import PARAMS, get_config 

39from netbox.object_actions import AddObject, BulkDelete, BulkExport, DeleteObject 

40from netbox.plugins import PluginConfig 

41from netbox.plugins.utils import get_installed_plugins 

42from netbox.ui import layout 

43from netbox.ui.breadcrumbs import Breadcrumb, filtered_list_url 

44from netbox.ui.panels import ( 

45 CommentsPanel, 

46 JSONPanel, 

47 ObjectsTablePanel, 

48 PluginContentPanel, 

49 RelatedObjectsPanel, 

50 TemplatePanel, 

51) 

52from netbox.views import generic 

53from netbox.views.generic.base import BaseObjectView 

54from netbox.views.generic.mixins import TableMixin 

55from utilities.apps import get_installed_apps 

56from utilities.data import deep_compare_dict 

57from utilities.forms import ConfirmationForm 

58from utilities.htmx import htmx_partial 

59from utilities.json import ConfigJSONEncoder 

60from utilities.query import count_related 

61from utilities.rqworker import get_all_workers 

62from utilities.views import ( 

63 ContentTypePermissionRequiredMixin, 

64 GetRelatedModelsMixin, 

65 GetReturnURLMixin, 

66 ViewTab, 

67 register_model_view, 

68) 

69 

70from . import filtersets, forms, tables 

71from .jobs import SyncDataSourceJob 

72from .models import * 

73from .plugins import get_catalog_plugins, get_local_plugins 

74from .tables import CatalogPluginTable, JobLogEntryTable, PluginVersionTable 

75from .ui import panels 

76 

77# 

78# Data sources 

79# 

80 

81 

82@register_model_view(DataSource, 'list', path='', detail=False) 

83class DataSourceListView(generic.ObjectListView): 

84 queryset = DataSource.objects.annotate( 

85 file_count=count_related(DataFile, 'source') 

86 ) 

87 filterset = filtersets.DataSourceFilterSet 

88 filterset_form = forms.DataSourceFilterForm 

89 table = tables.DataSourceTable 

90 

91 

92@register_model_view(DataSource) 

93class DataSourceView(GetRelatedModelsMixin, generic.ObjectView): 

94 queryset = DataSource.objects.all() 

95 layout = layout.SimpleLayout( 

96 left_panels=[ 

97 panels.DataSourcePanel(), 

98 TagsPanel(), 

99 CommentsPanel(), 

100 ], 

101 right_panels=[ 

102 panels.DataSourceBackendPanel(), 

103 RelatedObjectsPanel(), 

104 CustomFieldsPanel(), 

105 ], 

106 bottom_panels=[ 

107 ObjectsTablePanel( 

108 model='core.DataFile', 

109 filters={'source_id': lambda ctx: ctx['object'].pk}, 

110 exclude_columns=['source'], 

111 ), 

112 ], 

113 ) 

114 

115 def get_extra_context(self, request, instance): 

116 return { 

117 'related_models': self.get_related_models(request, instance), 

118 } 

119 

120 

121@register_model_view(DataSource, 'sync') 

122class DataSourceSyncView(GetReturnURLMixin, BaseObjectView): 

123 queryset = DataSource.objects.all() 

124 

125 def get_required_permission(self): 

126 return 'core.sync_datasource' 

127 

128 def get(self, request, pk): 

129 # Redirect GET requests to the object view 

130 datasource = get_object_or_404(self.queryset, pk=pk) 

131 return redirect(datasource.get_absolute_url()) 

132 

133 def post(self, request, pk): 

134 datasource = get_object_or_404(self.queryset, pk=pk) 

135 # Enqueue the sync job 

136 job = SyncDataSourceJob.enqueue(instance=datasource, user=request.user) 

137 messages.success( 

138 request, 

139 _("Queued job #{id} to sync {datasource}").format(id=job.pk, datasource=datasource) 

140 ) 

141 return redirect(self.get_return_url(request, datasource)) 

142 

143 

144@register_model_view(DataSource, 'add', detail=False) 

145@register_model_view(DataSource, 'edit') 

146class DataSourceEditView(generic.ObjectEditView): 

147 queryset = DataSource.objects.all() 

148 form = forms.DataSourceForm 

149 

150 

151@register_model_view(DataSource, 'delete') 

152class DataSourceDeleteView(generic.ObjectDeleteView): 

153 queryset = DataSource.objects.all() 

154 

155 

156@register_model_view(DataSource, 'bulk_import', path='import', detail=False) 

157class DataSourceBulkImportView(generic.BulkImportView): 

158 queryset = DataSource.objects.all() 

159 model_form = forms.DataSourceImportForm 

160 

161 

162@register_model_view(DataSource, 'bulk_edit', path='edit', detail=False) 

163class DataSourceBulkEditView(generic.BulkEditView): 

164 queryset = DataSource.objects.annotate( 

165 count_files=count_related(DataFile, 'source') 

166 ) 

167 filterset = filtersets.DataSourceFilterSet 

168 table = tables.DataSourceTable 

169 form = forms.DataSourceBulkEditForm 

170 

171 

172@register_model_view(DataSource, 'bulk_rename', path='rename', detail=False) 

173class DataSourceBulkRenameView(generic.BulkRenameView): 

174 queryset = DataSource.objects.all() 

175 filterset = filtersets.DataSourceFilterSet 

176 

177 

178@register_model_view(DataSource, 'bulk_delete', path='delete', detail=False) 

179class DataSourceBulkDeleteView(generic.BulkDeleteView): 

180 queryset = DataSource.objects.annotate( 

181 count_files=count_related(DataFile, 'source') 

182 ) 

183 filterset = filtersets.DataSourceFilterSet 

184 table = tables.DataSourceTable 

185 

186 

187# 

188# Data files 

189# 

190 

191@register_model_view(DataFile, 'list', path='', detail=False) 

192class DataFileListView(generic.ObjectListView): 

193 queryset = DataFile.objects.defer('data') 

194 filterset = filtersets.DataFileFilterSet 

195 filterset_form = forms.DataFileFilterForm 

196 table = tables.DataFileTable 

197 actions = (BulkDelete,) 

198 

199 

200@register_model_view(DataFile) 

201@method_decorator(never_cache, name='dispatch') 

202class DataFileView(generic.ObjectView): 

203 template_name = 'generic/object.html' 

204 queryset = DataFile.objects.all() 

205 actions = (DeleteObject,) 

206 layout = layout.Layout( 

207 layout.Row( 

208 layout.Column( 

209 panels.DataFilePanel(), 

210 panels.DataFileContentPanel(), 

211 PluginContentPanel('left_page'), 

212 ), 

213 ), 

214 layout.Row( 

215 layout.Column( 

216 PluginContentPanel('full_width_page'), 

217 ), 

218 ), 

219 breadcrumbs=[ 

220 Breadcrumb('source', url=filtered_list_url('core:datafile_list', 'source_id')), 

221 ], 

222 ) 

223 

224 

225@register_model_view(DataFile, 'delete') 

226class DataFileDeleteView(generic.ObjectDeleteView): 

227 queryset = DataFile.objects.all() 

228 

229 

230@register_model_view(DataFile, 'bulk_delete', path='delete', detail=False) 

231class DataFileBulkDeleteView(generic.BulkDeleteView): 

232 queryset = DataFile.objects.defer('data') 

233 filterset = filtersets.DataFileFilterSet 

234 table = tables.DataFileTable 

235 

236 

237# 

238# Jobs 

239# 

240 

241@register_model_view(Job, 'list', path='', detail=False) 

242class JobListView(generic.ObjectListView): 

243 queryset = Job.objects.defer('data') 

244 filterset = filtersets.JobFilterSet 

245 filterset_form = forms.JobFilterForm 

246 table = tables.JobTable 

247 actions = (BulkExport, BulkDelete) 

248 

249 

250@register_model_view(Job) 

251class JobView(generic.ObjectView): 

252 queryset = Job.objects.all() 

253 actions = (DeleteObject,) 

254 layout = layout.SimpleLayout( 

255 left_panels=[ 

256 panels.JobPanel(), 

257 ], 

258 right_panels=[ 

259 panels.JobSchedulingPanel(), 

260 ], 

261 bottom_panels=[ 

262 JSONPanel('data', title=_('Data')), 

263 ], 

264 ) 

265 

266 

267@register_model_view(Job, 'log') 

268class JobLogView(generic.ObjectView): 

269 queryset = Job.objects.all() 

270 actions = (DeleteObject,) 

271 template_name = 'core/job/log.html' 

272 tab = ViewTab( 

273 label=_('Log'), 

274 badge=lambda obj: len(obj.log_entries), 

275 weight=500, 

276 ) 

277 layout = layout.Layout( 

278 layout.Row( 

279 layout.Column( 

280 TemplatePanel('core/job/log_entries.html', title=_('Log Entries')), 

281 PluginContentPanel('left_page'), 

282 ), 

283 ), 

284 layout.Row( 

285 layout.Column( 

286 PluginContentPanel('full_width_page'), 

287 ), 

288 ), 

289 ) 

290 

291 def get_table(self, request, instance): 

292 table = JobLogEntryTable(instance.log_entries) 

293 table.embedded = True 

294 table.htmx_url = reverse('core:job_log', kwargs={'pk': instance.pk}) 

295 table.configure(request) 

296 return table 

297 

298 def get_extra_context(self, request, instance): 

299 return { 

300 'table': self.get_table(request, instance), 

301 } 

302 

303 def get(self, request, **kwargs): 

304 if htmx_partial(request): 

305 instance = self.get_object(**kwargs) 

306 return render(request, 'htmx/table.html', { 

307 'object': instance, 

308 'table': self.get_table(request, instance), 

309 }) 

310 return super().get(request, **kwargs) 

311 

312 

313@register_model_view(Job, 'delete') 

314class JobDeleteView(generic.ObjectDeleteView): 

315 queryset = Job.objects.defer('data') 

316 

317 

318@register_model_view(Job, 'bulk_delete', path='delete', detail=False) 

319class JobBulkDeleteView(generic.BulkDeleteView): 

320 queryset = Job.objects.defer('data') 

321 filterset = filtersets.JobFilterSet 

322 table = tables.JobTable 

323 

324 

325# 

326# Change logging 

327# 

328 

329@register_model_view(ObjectChange, 'list', path='', detail=False) 

330class ObjectChangeListView(generic.ObjectListView): 

331 queryset = None 

332 filterset = filtersets.ObjectChangeFilterSet 

333 filterset_form = forms.ObjectChangeFilterForm 

334 table = tables.ObjectChangeTable 

335 template_name = 'core/objectchange_list.html' 

336 actions = (BulkExport,) 

337 

338 def get_queryset(self, request): 

339 return ObjectChange.objects.valid_models() 

340 

341 

342@register_model_view(ObjectChange) 

343class ObjectChangeView(generic.ObjectView): 

344 queryset = None 

345 layout = layout.Layout( 

346 layout.Row( 

347 layout.Column(panels.ObjectChangePanel()), 

348 layout.Column(TemplatePanel('core/panels/objectchange_difference.html')), 

349 ), 

350 layout.Row( 

351 layout.Column(TemplatePanel('core/panels/objectchange_prechange.html')), 

352 layout.Column(TemplatePanel('core/panels/objectchange_postchange.html')), 

353 ), 

354 layout.Row( 

355 layout.Column(PluginContentPanel('left_page')), 

356 layout.Column(PluginContentPanel('right_page')), 

357 ), 

358 layout.Row( 

359 layout.Column( 

360 TemplatePanel('core/panels/objectchange_related.html'), 

361 PluginContentPanel('full_width_page'), 

362 ), 

363 ), 

364 ) 

365 

366 def get_queryset(self, request): 

367 return ObjectChange.objects.valid_models() 

368 

369 def get_extra_context(self, request, instance): 

370 related_changes = ObjectChange.objects.valid_models().restrict(request.user, 'view').filter( 

371 request_id=instance.request_id 

372 ).exclude( 

373 pk=instance.pk 

374 ) 

375 related_changes_table = tables.ObjectChangeTable( 

376 data=related_changes[:50], 

377 orderable=False 

378 ) 

379 related_changes_table.configure(request) 

380 

381 objectchanges = ObjectChange.objects.valid_models().restrict(request.user, 'view').filter( 

382 changed_object_type=instance.changed_object_type, 

383 changed_object_id=instance.changed_object_id, 

384 ) 

385 

386 next_change = objectchanges.filter(time__gt=instance.time).order_by('time').first() 

387 prev_change = objectchanges.filter(time__lt=instance.time).order_by('-time').first() 

388 

389 if not instance.prechange_data and instance.action in ['update', 'delete'] and prev_change: 

390 non_atomic_change = True 

391 prechange_data = prev_change.postchange_data_clean 

392 else: 

393 non_atomic_change = False 

394 prechange_data = instance.prechange_data_clean 

395 

396 if prechange_data and instance.postchange_data: 

397 diff_added, diff_removed = deep_compare_dict( 

398 prechange_data, 

399 instance.postchange_data_clean, 

400 exclude=['last_updated'], 

401 ) 

402 else: 

403 diff_added = {} 

404 diff_removed = {} 

405 

406 return { 

407 'diff_added': diff_added, 

408 'diff_removed': diff_removed, 

409 'next_change': next_change, 

410 'prev_change': prev_change, 

411 'related_changes_table': related_changes_table, 

412 'related_changes_count': related_changes.count(), 

413 'non_atomic_change': non_atomic_change 

414 } 

415 

416 

417# 

418# Config Revisions 

419# 

420 

421@register_model_view(ConfigRevision, 'list', path='', detail=False) 

422class ConfigRevisionListView(generic.ObjectListView): 

423 queryset = ConfigRevision.objects.all() 

424 filterset = filtersets.ConfigRevisionFilterSet 

425 filterset_form = forms.ConfigRevisionFilterForm 

426 table = tables.ConfigRevisionTable 

427 actions = (AddObject, BulkExport) 

428 

429 

430@register_model_view(ConfigRevision) 

431class ConfigRevisionView(generic.ObjectView): 

432 queryset = ConfigRevision.objects.all() 

433 layout = layout.Layout( 

434 layout.Row( 

435 layout.Column( 

436 TemplatePanel('core/panels/configrevision_data.html'), 

437 TemplatePanel('core/panels/configrevision_comment.html'), 

438 PluginContentPanel('left_page'), 

439 ), 

440 ), 

441 layout.Row( 

442 layout.Column( 

443 PluginContentPanel('full_width_page'), 

444 ), 

445 ), 

446 root_breadcrumb=False, 

447 ) 

448 

449 def get_extra_context(self, request, instance): 

450 """ 

451 Retrieve additional context for a given request and instance. 

452 """ 

453 # Copy the revision data to avoid modifying the original 

454 config = deepcopy(instance.data or {}) 

455 

456 # Serialize any JSON-based classes 

457 for attr in ['CUSTOM_VALIDATORS', 'DEFAULT_USER_PREFERENCES', 'PROTECTION_RULES']: 

458 if attr in config: 

459 config[attr] = json.dumps(config[attr], cls=ConfigJSONEncoder, indent=4) 

460 

461 return { 

462 'config': config, 

463 } 

464 

465 

466@register_model_view(ConfigRevision, 'add', detail=False) 

467class ConfigRevisionEditView(generic.ObjectEditView): 

468 queryset = ConfigRevision.objects.all() 

469 form = forms.ConfigRevisionForm 

470 

471 

472@register_model_view(ConfigRevision, 'delete') 

473class ConfigRevisionDeleteView(generic.ObjectDeleteView): 

474 queryset = ConfigRevision.objects.all() 

475 

476 

477@register_model_view(ConfigRevision, 'bulk_delete', path='delete', detail=False) 

478class ConfigRevisionBulkDeleteView(generic.BulkDeleteView): 

479 queryset = ConfigRevision.objects.all() 

480 filterset = filtersets.ConfigRevisionFilterSet 

481 table = tables.ConfigRevisionTable 

482 

483 

484@register_model_view(ConfigRevision, 'restore') 

485class ConfigRevisionRestoreView(ContentTypePermissionRequiredMixin, View): 

486 

487 def get_required_permission(self): 

488 return 'core.configrevision_edit' 

489 

490 def get(self, request, pk): 

491 candidate_config = get_object_or_404(ConfigRevision, pk=pk) 

492 

493 # Get the current ConfigRevision 

494 config_version = get_config().version 

495 current_config = ConfigRevision.objects.filter(pk=config_version).first() 

496 

497 params = [] 

498 for param in PARAMS: 

499 params.append(( 

500 param.name, 

501 current_config.data.get(param.name, None) if current_config else None, 

502 candidate_config.data.get(param.name, None) 

503 )) 

504 

505 return render(request, 'core/configrevision_restore.html', { 

506 'object': candidate_config, 

507 'params': params, 

508 }) 

509 

510 def post(self, request, pk): 

511 if not request.user.has_perm('core.configrevision_edit'): 

512 return HttpResponseForbidden() 

513 

514 candidate_config = get_object_or_404(ConfigRevision, pk=pk) 

515 candidate_config.activate() 

516 messages.success(request, _("Restored configuration revision #{id}").format(id=pk)) 

517 

518 return redirect(candidate_config.get_absolute_url()) 

519 

520 

521# 

522# Background Tasks (RQ) 

523# 

524 

525class BaseRQView(UserPassesTestMixin, View): 

526 

527 def test_func(self): 

528 return self.request.user.is_superuser 

529 

530 

531class BackgroundQueueListView(TableMixin, BaseRQView): 

532 table = tables.BackgroundQueueTable 

533 

534 def get(self, request): 

535 data = get_statistics(run_maintenance_tasks=True)["queues"] 

536 table = self.get_table(data, request, bulk_actions=False) 

537 

538 return render(request, 'core/rq_queue_list.html', { 

539 'table': table, 

540 }) 

541 

542 

543class BackgroundTaskListView(TableMixin, BaseRQView): 

544 table = tables.BackgroundTaskTable 

545 

546 def get_table_data(self, request, queue, status): 

547 

548 # Call get_jobs() to returned queued tasks 

549 if status == RQJobStatus.QUEUED: 

550 return queue.get_jobs() 

551 

552 return get_rq_jobs_from_status(queue, status) 

553 

554 def get(self, request, queue_index, status): 

555 queue = get_queue_by_index(queue_index) 

556 data = self.get_table_data(request, queue, status) 

557 table = self.get_table(data, request, False) 

558 

559 # If this is an HTMX request, return only the rendered table HTML 

560 if htmx_partial(request): 

561 return render(request, 'htmx/table.html', { 

562 'table': table, 

563 }) 

564 

565 return render(request, 'core/rq_task_list.html', { 

566 'table': table, 

567 'queue': queue, 

568 'status': status, 

569 }) 

570 

571 

572class BackgroundTaskView(BaseRQView): 

573 

574 def get(self, request, job_id): 

575 # all the RQ queues should use the same connection 

576 config = get_queues_list()[0] 

577 try: 

578 job = RQ_Job.fetch(job_id, connection=get_redis_connection(config['connection_config']),) 

579 except NoSuchJobError: 

580 raise Http404(_("Job {job_id} not found").format(job_id=job_id)) 

581 

582 queue_index = get_queues_map()[job.origin] 

583 queue = get_queue_by_index(queue_index) 

584 

585 try: 

586 exc_info = job._exc_info 

587 except AttributeError: 

588 exc_info = None 

589 

590 return render(request, 'core/rq_task.html', { 

591 'queue': queue, 

592 'job': job, 

593 'queue_index': queue_index, 

594 'dependency_id': job._dependency_id, 

595 'exc_info': exc_info, 

596 }) 

597 

598 

599class BackgroundTaskDeleteView(BaseRQView): 

600 

601 def get(self, request, job_id): 

602 if not request.htmx: 

603 return redirect(reverse('core:background_queue_list')) 

604 

605 form = ConfirmationForm(initial=request.GET) 

606 

607 return render(request, 'htmx/delete_form.html', { 

608 'object_type': 'background task', 

609 'object': job_id, 

610 'form': form, 

611 'form_url': reverse('core:background_task_delete', kwargs={'job_id': job_id}) 

612 }) 

613 

614 def post(self, request, job_id): 

615 form = ConfirmationForm(request.POST) 

616 

617 if form.is_valid(): 

618 delete_rq_job(job_id) 

619 messages.success(request, _('Job {id} has been deleted.').format(id=job_id)) 

620 else: 

621 messages.error(request, _('Error deleting job {id}: {error}').format(id=job_id, error=form.errors[0])) 

622 

623 return redirect(reverse('core:background_queue_list')) 

624 

625 

626class BackgroundTaskRequeueView(BaseRQView): 

627 

628 def get(self, request, job_id): 

629 requeue_rq_job(job_id) 

630 messages.success(request, _('Job {id} has been re-enqueued.').format(id=job_id)) 

631 return redirect(reverse('core:background_task', args=[job_id])) 

632 

633 

634class BackgroundTaskEnqueueView(BaseRQView): 

635 

636 def get(self, request, job_id): 

637 # all the RQ queues should use the same connection 

638 enqueue_rq_job(job_id) 

639 messages.success(request, _('Job {id} has been enqueued.').format(id=job_id)) 

640 return redirect(reverse('core:background_task', args=[job_id])) 

641 

642 

643class BackgroundTaskStopView(BaseRQView): 

644 

645 def get(self, request, job_id): 

646 stopped_jobs = stop_rq_job(job_id) 

647 if len(stopped_jobs) == 1: 

648 messages.success(request, _('Job {id} has been stopped.').format(id=job_id)) 

649 else: 

650 messages.error(request, _('Failed to stop job {id}').format(id=job_id)) 

651 

652 return redirect(reverse('core:background_task', args=[job_id])) 

653 

654 

655class WorkerListView(TableMixin, BaseRQView): 

656 table = tables.WorkerTable 

657 

658 def get_table_data(self, request, queue): 

659 clean_worker_registry(queue) 

660 all_workers = Worker.all(queue.connection) 

661 workers = [worker for worker in all_workers if queue.name in worker.queue_names()] 

662 return workers 

663 

664 def get(self, request, queue_index): 

665 queue = get_queue_by_index(queue_index) 

666 data = self.get_table_data(request, queue) 

667 

668 table = self.get_table(data, request, False) 

669 

670 # If this is an HTMX request, return only the rendered table HTML 

671 if htmx_partial(request): 

672 if not request.htmx.target: 

673 table.embedded = True 

674 # Hide selection checkboxes 

675 if 'pk' in table.base_columns: 

676 table.columns.hide('pk') 

677 return render(request, 'htmx/table.html', { 

678 'table': table, 

679 'queue': queue, 

680 }) 

681 

682 return render(request, 'core/rq_worker_list.html', { 

683 'table': table, 

684 'queue': queue, 

685 }) 

686 

687 

688class WorkerView(BaseRQView): 

689 

690 def get(self, request, key): 

691 # all the RQ queues should use the same connection 

692 config = get_queues_list()[0] 

693 worker = Worker.find_by_key('rq:worker:' + key, connection=get_redis_connection(config['connection_config'])) 

694 # Convert microseconds to milliseconds 

695 worker.total_working_time = worker.total_working_time / 1000 

696 

697 return render(request, 'core/rq_worker.html', { 

698 'worker': worker, 

699 'job': worker.get_current_job(), 

700 'total_working_time': worker.total_working_time * 1000, 

701 }) 

702 

703 

704# 

705# System 

706# 

707 

708 

709class SystemView(UserPassesTestMixin, View): 

710 

711 def test_func(self): 

712 return self.request.user.is_superuser 

713 

714 def _get_stats(self): 

715 psql_version = db_name = db_size = None 

716 try: 

717 with connection.cursor() as cursor: 

718 cursor.execute("SELECT version()") 

719 psql_version = cursor.fetchone()[0] 

720 psql_version = psql_version.split('(')[0].strip() 

721 cursor.execute("SELECT current_database()") 

722 db_name = cursor.fetchone()[0] 

723 cursor.execute("SELECT pg_size_pretty(pg_database_size(current_database()))") 

724 db_size = cursor.fetchone()[0] 

725 except (DatabaseError, IndexError): 

726 pass 

727 return { 

728 'netbox_release': settings.RELEASE, 

729 'django_version': django_version, 

730 'python_version': platform.python_version(), 

731 'postgresql_version': psql_version, 

732 'database_name': db_name, 

733 'database_size': db_size, 

734 'rq_worker_count': len(get_all_workers()), 

735 } 

736 

737 def _get_object_counts(self): 

738 objects = {} 

739 for ot in ObjectType.objects.public().order_by('app_label', 'model'): 

740 if model := ot.model_class(): 

741 objects[ot] = model.objects.count() 

742 return objects 

743 

744 def get(self, request): 

745 stats = self._get_stats() 

746 django_apps = get_installed_apps() 

747 config = get_config() 

748 plugins = get_installed_plugins() 

749 objects = self._get_object_counts() 

750 

751 # Raw data export 

752 if 'export' in request.GET: 

753 db_schema = get_db_schema() 

754 stats['netbox_release'] = stats['netbox_release'].asdict() 

755 params = [param.name for param in PARAMS] 

756 data = { 

757 **stats, 

758 'django_apps': django_apps, 

759 'plugins': plugins, 

760 'config': { 

761 k: getattr(config, k) for k in sorted(params) 

762 }, 

763 'objects': { 

764 f'{ot.app_label}.{ot.model}': count for ot, count in objects.items() 

765 }, 

766 'db_schema': { 

767 table['name']: { 

768 'columns': table['columns'], 

769 'indexes': table['indexes'], 

770 } for table in db_schema 

771 }, 

772 } 

773 response = HttpResponse(json.dumps(data, cls=ConfigJSONEncoder, indent=4), content_type='text/json') 

774 response['Content-Disposition'] = content_disposition_header(as_attachment=True, filename='netbox.json') 

775 return response 

776 

777 # Serialize any JSON-based classes 

778 for attr in ['CUSTOM_VALIDATORS', 'DEFAULT_USER_PREFERENCES', 'PROTECTION_RULES']: 

779 if hasattr(config, attr) and getattr(config, attr, None): 

780 setattr(config, attr, json.dumps(getattr(config, attr), cls=ConfigJSONEncoder, indent=4)) 

781 

782 return render(request, 'core/system.html', { 

783 'stats': stats, 

784 'django_apps': django_apps, 

785 'config': config, 

786 'plugins': plugins,