Coverage for dcim/api/views.py: 80%
320 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1from django.contrib.contenttypes.prefetch import GenericPrefetch
2from django.http import Http404, HttpResponse
3from django.shortcuts import get_object_or_404
4from drf_spectacular.types import OpenApiTypes
5from drf_spectacular.utils import OpenApiParameter, extend_schema
6from rest_framework.decorators import action
7from rest_framework.response import Response
8from rest_framework.routers import APIRootView
9from rest_framework.viewsets import ViewSet
11from dcim import filtersets
12from dcim.constants import CABLE_TRACE_SVG_DEFAULT_WIDTH
13from dcim.models import *
14from dcim.svg import CableTraceSVG
15from extras.api.mixins import ConfigContextQuerySetMixin, RenderConfigMixin
16from netbox.api.authentication import IsAuthenticatedOrLoginNotRequired
17from netbox.api.metadata import ContentTypeMetadata
18from netbox.api.pagination import StripCountAnnotationsPaginator
19from netbox.api.viewsets import NetBoxModelViewSet, NetBoxReadOnlyModelViewSet
20from utilities.api import get_serializer_for_model
21from utilities.query import count_related
22from utilities.query_functions import CollateAsChar
23from virtualization.models import VirtualMachine
25from . import serializers
26from .exceptions import MissingFilterException
29class DCIMRootView(APIRootView):
30 """
31 DCIM API root view
32 """
33 def get_view_name(self):
34 return 'DCIM'
37# Mixins
39class PathEndpointMixin:
41 @action(detail=True, url_path='trace')
42 def trace(self, request, pk):
43 """
44 Trace a complete cable path and return each segment as a three-tuple of (termination, cable, termination).
45 """
46 obj = get_object_or_404(self.queryset, pk=pk)
48 # Initialize the path array
49 path = []
51 # Render SVG image if requested
52 if request.GET.get('render', None) == 'svg': 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true
53 try:
54 width = int(request.GET.get('width', CABLE_TRACE_SVG_DEFAULT_WIDTH))
55 except (ValueError, TypeError):
56 width = CABLE_TRACE_SVG_DEFAULT_WIDTH
57 drawing = CableTraceSVG(obj, base_url=request.build_absolute_uri('/'), width=width)
58 return HttpResponse(drawing.render().tostring(), content_type='image/svg+xml')
60 # Serialize path objects, iterating over each three-tuple in the path
61 for near_ends, cable, far_ends in obj.trace(): 61 ↛ 62line 61 didn't jump to line 62 because the loop on line 61 never started
62 if near_ends:
63 serializer_a = get_serializer_for_model(near_ends[0])
64 near_ends = serializer_a(near_ends, nested=True, many=True, context={'request': request}).data
65 else:
66 # Path is split; stop here
67 break
68 if cable:
69 cable = serializers.TracedCableSerializer(cable[0], context={'request': request}).data
70 if far_ends:
71 serializer_b = get_serializer_for_model(far_ends[0])
72 far_ends = serializer_b(far_ends, nested=True, many=True, context={'request': request}).data
74 path.append((near_ends, cable, far_ends))
76 return Response(path)
79class PassThroughPortMixin:
81 @action(detail=True, url_path='paths')
82 def paths(self, request, pk):
83 """
84 Return all CablePaths which traverse a given pass-through port.
85 """
86 obj = get_object_or_404(self.queryset, pk=pk)
87 cablepaths = CablePath.objects.filter(_nodes__contains=obj)
88 serializer = serializers.CablePathSerializer(cablepaths, context={'request': request}, many=True)
90 return Response(serializer.data)
93#
94# Regions
95#
97class RegionViewSet(NetBoxModelViewSet):
98 queryset = Region.objects.add_related_count(
99 Region.objects.all(),
100 Site,
101 'region',
102 'site_count',
103 cumulative=True
104 )
105 serializer_class = serializers.RegionSerializer
106 filterset_class = filtersets.RegionFilterSet
109#
110# Site groups
111#
113class SiteGroupViewSet(NetBoxModelViewSet):
114 queryset = SiteGroup.objects.add_related_count(
115 SiteGroup.objects.all(),
116 Site,
117 'group',
118 'site_count',
119 cumulative=True
120 )
121 serializer_class = serializers.SiteGroupSerializer
122 filterset_class = filtersets.SiteGroupFilterSet
125#
126# Sites
127#
129class SiteViewSet(NetBoxModelViewSet):
130 queryset = Site.objects.all()
131 serializer_class = serializers.SiteSerializer
132 filterset_class = filtersets.SiteFilterSet
135#
136# Locations
137#
139class LocationViewSet(NetBoxModelViewSet):
140 queryset = Location.objects.add_related_count(
141 Location.objects.add_related_count(
142 Location.objects.all(),
143 Device,
144 'location',
145 'device_count',
146 cumulative=True
147 ),
148 Rack,
149 'location',
150 'rack_count',
151 cumulative=True
152 )
153 serializer_class = serializers.LocationSerializer
154 filterset_class = filtersets.LocationFilterSet
157#
158# Rack groups
159#
162class RackGroupViewSet(NetBoxModelViewSet):
163 queryset = RackGroup.objects.all()
164 serializer_class = serializers.RackGroupSerializer
165 filterset_class = filtersets.RackGroupFilterSet
168#
169# Rack roles
170#
172class RackRoleViewSet(NetBoxModelViewSet):
173 queryset = RackRole.objects.all()
174 serializer_class = serializers.RackRoleSerializer
175 filterset_class = filtersets.RackRoleFilterSet
178#
179# Rack Types
180#
182class RackTypeViewSet(NetBoxModelViewSet):
183 queryset = RackType.objects.all()
184 serializer_class = serializers.RackTypeSerializer
185 filterset_class = filtersets.RackTypeFilterSet
188#
189# Racks
190#
192class RackViewSet(NetBoxModelViewSet):
193 queryset = Rack.objects.all()
194 serializer_class = serializers.RackSerializer
195 filterset_class = filtersets.RackFilterSet
197 @extend_schema(
198 operation_id='dcim_racks_elevation_retrieve',
199 filters=False,
200 parameters=[serializers.RackElevationDetailFilterSerializer],
201 responses={200: serializers.RackUnitSerializer(many=True)}
202 )
203 @action(detail=True)
204 def elevation(self, request, pk=None):
205 """
206 Rack elevation representing the list of rack units. Also supports rendering the elevation as an SVG.
207 """
208 rack = get_object_or_404(self.queryset, pk=pk)
209 serializer = serializers.RackElevationDetailFilterSerializer(data=request.GET)
210 if not serializer.is_valid():
211 return Response(serializer.errors, 400)
212 data = serializer.validated_data
214 if data['render'] == 'svg':
215 # Determine attributes for highlighting devices (if any)
216 highlight_params = []
217 for param in request.GET.getlist('highlight'):
218 try:
219 highlight_params.append(param.split(':', 1))
220 except ValueError:
221 pass
223 # Render and return the elevation as an SVG drawing with the correct content type
224 drawing = rack.get_elevation_svg(
225 face=data['face'],
226 user=request.user,
227 unit_width=data['unit_width'],
228 unit_height=data['unit_height'],
229 legend_width=data['legend_width'],
230 include_images=data['include_images'],
231 base_url=request.build_absolute_uri('/'),
232 highlight_params=highlight_params
233 )
234 return HttpResponse(drawing.tostring(), content_type='image/svg+xml')
236 # Return a JSON representation of the rack units in the elevation
237 elevation = rack.get_rack_units(
238 face=data['face'],
239 user=request.user,
240 exclude=data['exclude'],
241 expand_devices=data['expand_devices']
242 )
244 # Enable filtering rack units by ID
245 if q := data['q']:
246 q = q.lower()
247 elevation = [u for u in elevation if q in str(u['id']) or q in str(u['name']).lower()]
249 page = self.paginate_queryset(elevation)
250 if page is not None:
251 rack_units = serializers.RackUnitSerializer(page, many=True, context={'request': request})
252 return self.get_paginated_response(rack_units.data)
254 # TODO: This endpoint should always return an HttpResponse/DRF Response; `None` is not a meaningful result.
255 return None
258#
259# Rack reservations
260#
262class RackReservationViewSet(NetBoxModelViewSet):
263 queryset = RackReservation.objects.all()
264 serializer_class = serializers.RackReservationSerializer
265 filterset_class = filtersets.RackReservationFilterSet
268#
269# Manufacturers
270#
272class ManufacturerViewSet(NetBoxModelViewSet):
273 queryset = Manufacturer.objects.all()
274 serializer_class = serializers.ManufacturerSerializer
275 filterset_class = filtersets.ManufacturerFilterSet
278#
279# Device/module types
280#
282class DeviceTypeViewSet(NetBoxModelViewSet):
283 queryset = DeviceType.objects.all()
284 serializer_class = serializers.DeviceTypeSerializer
285 filterset_class = filtersets.DeviceTypeFilterSet
288class ModuleBayTypeViewSet(NetBoxModelViewSet):
289 queryset = ModuleBayType.objects.all()
290 serializer_class = serializers.ModuleBayTypeSerializer
291 filterset_class = filtersets.ModuleBayTypeFilterSet
294class ModuleTypeProfileViewSet(NetBoxModelViewSet):
295 queryset = ModuleTypeProfile.objects.all()
296 serializer_class = serializers.ModuleTypeProfileSerializer
297 filterset_class = filtersets.ModuleTypeProfileFilterSet
300class ModuleTypeViewSet(NetBoxModelViewSet):
301 queryset = ModuleType.objects.prefetch_related('module_bay_types__manufacturer')
302 serializer_class = serializers.ModuleTypeSerializer
303 filterset_class = filtersets.ModuleTypeFilterSet
306#
307# Device type components
308#
310class ConsolePortTemplateViewSet(NetBoxModelViewSet):
311 queryset = ConsolePortTemplate.objects.all()
312 serializer_class = serializers.ConsolePortTemplateSerializer
313 filterset_class = filtersets.ConsolePortTemplateFilterSet
316class ConsoleServerPortTemplateViewSet(NetBoxModelViewSet):
317 queryset = ConsoleServerPortTemplate.objects.all()
318 serializer_class = serializers.ConsoleServerPortTemplateSerializer
319 filterset_class = filtersets.ConsoleServerPortTemplateFilterSet
322class PowerPortTemplateViewSet(NetBoxModelViewSet):
323 queryset = PowerPortTemplate.objects.all()
324 serializer_class = serializers.PowerPortTemplateSerializer
325 filterset_class = filtersets.PowerPortTemplateFilterSet
328class PowerOutletTemplateViewSet(NetBoxModelViewSet):
329 queryset = PowerOutletTemplate.objects.all()
330 serializer_class = serializers.PowerOutletTemplateSerializer
331 filterset_class = filtersets.PowerOutletTemplateFilterSet
334class CoolingIntakeTemplateViewSet(NetBoxModelViewSet):
335 queryset = CoolingIntakeTemplate.objects.all()
336 serializer_class = serializers.CoolingIntakeTemplateSerializer
337 filterset_class = filtersets.CoolingIntakeTemplateFilterSet
340class CoolingOutflowTemplateViewSet(NetBoxModelViewSet):
341 queryset = CoolingOutflowTemplate.objects.all()
342 serializer_class = serializers.CoolingOutflowTemplateSerializer
343 filterset_class = filtersets.CoolingOutflowTemplateFilterSet
346class InterfaceTemplateViewSet(NetBoxModelViewSet):
347 queryset = InterfaceTemplate.objects.all()
348 serializer_class = serializers.InterfaceTemplateSerializer
349 filterset_class = filtersets.InterfaceTemplateFilterSet
352class FrontPortTemplateViewSet(NetBoxModelViewSet):
353 queryset = FrontPortTemplate.objects.all()
354 serializer_class = serializers.FrontPortTemplateSerializer
355 filterset_class = filtersets.FrontPortTemplateFilterSet
358class RearPortTemplateViewSet(NetBoxModelViewSet):
359 queryset = RearPortTemplate.objects.all()
360 serializer_class = serializers.RearPortTemplateSerializer
361 filterset_class = filtersets.RearPortTemplateFilterSet
364class ModuleBayTemplateViewSet(NetBoxModelViewSet):
365 queryset = ModuleBayTemplate.objects.prefetch_related('module_bay_types__manufacturer')
366 serializer_class = serializers.ModuleBayTemplateSerializer
367 filterset_class = filtersets.ModuleBayTemplateFilterSet
370class DeviceBayTemplateViewSet(NetBoxModelViewSet):
371 queryset = DeviceBayTemplate.objects.all()
372 serializer_class = serializers.DeviceBayTemplateSerializer
373 filterset_class = filtersets.DeviceBayTemplateFilterSet
376class InventoryItemTemplateViewSet(NetBoxModelViewSet):
377 queryset = InventoryItemTemplate.objects.all()
378 serializer_class = serializers.InventoryItemTemplateSerializer
379 filterset_class = filtersets.InventoryItemTemplateFilterSet
382#
383# Device roles
384#
386class DeviceRoleViewSet(NetBoxModelViewSet):
387 queryset = DeviceRole.objects.add_related_count(
388 DeviceRole.objects.add_related_count(
389 DeviceRole.objects.all(),
390 VirtualMachine,
391 'role',
392 'virtualmachine_count',
393 cumulative=True
394 ),
395 Device,
396 'role',
397 'device_count',
398 cumulative=True
399 )
400 serializer_class = serializers.DeviceRoleSerializer
401 filterset_class = filtersets.DeviceRoleFilterSet
404#
405# Platforms
406#
408class PlatformViewSet(NetBoxModelViewSet):
409 queryset = Platform.objects.add_related_count(
410 Platform.objects.add_related_count(
411 Platform.objects.all(),
412 VirtualMachine,
413 'platform',
414 'virtualmachine_count',
415 cumulative=True
416 ),
417 Device,
418 'platform',
419 'device_count',
420 cumulative=True
421 )
422 serializer_class = serializers.PlatformSerializer
423 filterset_class = filtersets.PlatformFilterSet
426#
427# Devices/modules
428#
430class DeviceViewSet(ConfigContextQuerySetMixin, RenderConfigMixin, NetBoxModelViewSet):
431 queryset = Device.objects.prefetch_related(
432 'device_type__manufacturer', # Referenced by Device.__str__() for unnamed devices
433 'parent_bay', # Referenced by DeviceSerializer.get_parent_device()
434 )
435 serializer_class = serializers.DeviceSerializer
436 filterset_class = filtersets.DeviceFilterSet
437 pagination_class = StripCountAnnotationsPaginator
440class VirtualDeviceContextViewSet(NetBoxModelViewSet):
441 queryset = VirtualDeviceContext.objects.all()
442 serializer_class = serializers.VirtualDeviceContextSerializer
443 filterset_class = filtersets.VirtualDeviceContextFilterSet
446class ModuleViewSet(NetBoxModelViewSet):
447 queryset = Module.objects.prefetch_related(
448 'module_bay__module_bay_types',
449 'module_type__module_bay_types',
450 )
451 serializer_class = serializers.ModuleSerializer
452 filterset_class = filtersets.ModuleFilterSet
455#
456# Device components
457#
459class ConsolePortViewSet(PathEndpointMixin, NetBoxModelViewSet):
460 queryset = ConsolePort.objects.prefetch_related(
461 '_path', 'cable__terminations',
462 )
463 serializer_class = serializers.ConsolePortSerializer
464 filterset_class = filtersets.ConsolePortFilterSet
467class ConsoleServerPortViewSet(PathEndpointMixin, NetBoxModelViewSet):
468 queryset = ConsoleServerPort.objects.prefetch_related(
469 '_path', 'cable__terminations',
470 )
471 serializer_class = serializers.ConsoleServerPortSerializer
472 filterset_class = filtersets.ConsoleServerPortFilterSet
475class PowerPortViewSet(PathEndpointMixin, NetBoxModelViewSet):
476 queryset = PowerPort.objects.prefetch_related(
477 '_path', 'cable__terminations',
478 )
479 serializer_class = serializers.PowerPortSerializer
480 filterset_class = filtersets.PowerPortFilterSet
483class PowerOutletViewSet(PathEndpointMixin, NetBoxModelViewSet):
484 queryset = PowerOutlet.objects.prefetch_related(
485 '_path', 'cable__terminations',
486 )
487 serializer_class = serializers.PowerOutletSerializer
488 filterset_class = filtersets.PowerOutletFilterSet
491class CoolingIntakeViewSet(NetBoxModelViewSet):
492 queryset = CoolingIntake.objects.all()
493 serializer_class = serializers.CoolingIntakeSerializer
494 filterset_class = filtersets.CoolingIntakeFilterSet
497class CoolingOutflowViewSet(NetBoxModelViewSet):
498 queryset = CoolingOutflow.objects.all()
499 serializer_class = serializers.CoolingOutflowSerializer
500 filterset_class = filtersets.CoolingOutflowFilterSet
503class InterfaceViewSet(PathEndpointMixin, NetBoxModelViewSet):
504 queryset = Interface.objects.prefetch_related(
505 GenericPrefetch(
506 "cable__terminations__termination",
507 [
508 Interface.objects.select_related("device", "cable"),
509 ],
510 ),
511 GenericPrefetch(
512 "_path__path_objects",
513 [
514 Interface.objects.select_related("device", "cable"),
515 ],
516 ),
517 'virtual_circuit_termination',
518 'l2vpn_terminations', # Referenced by InterfaceSerializer.l2vpn_termination
519 'ip_addresses', # Referenced by Interface.count_ipaddresses()
520 'fhrp_group_assignments', # Referenced by Interface.count_fhrp_groups()
521 )
522 serializer_class = serializers.InterfaceSerializer
523 filterset_class = filtersets.InterfaceFilterSet
525 def get_bulk_destroy_queryset(self):
526 # Ensure child interfaces are deleted prior to their parents
527 return self.get_queryset().order_by('device', 'parent', CollateAsChar('_name'))
530class FrontPortViewSet(PassThroughPortMixin, NetBoxModelViewSet):
531 queryset = FrontPort.objects.prefetch_related(
532 'cable__terminations',
533 )
534 serializer_class = serializers.FrontPortSerializer
535 filterset_class = filtersets.FrontPortFilterSet
538class RearPortViewSet(PassThroughPortMixin, NetBoxModelViewSet):
539 queryset = RearPort.objects.prefetch_related(
540 'cable__terminations',
541 )
542 serializer_class = serializers.RearPortSerializer
543 filterset_class = filtersets.RearPortFilterSet
546class ModuleBayViewSet(NetBoxModelViewSet):
547 queryset = ModuleBay.objects.prefetch_related(
548 'module_bay_types__manufacturer',
549 'installed_module__module_type__module_bay_types',
550 )
551 serializer_class = serializers.ModuleBaySerializer
552 filterset_class = filtersets.ModuleBayFilterSet
555class DeviceBayViewSet(NetBoxModelViewSet):
556 queryset = DeviceBay.objects.all()
557 serializer_class = serializers.DeviceBaySerializer
558 filterset_class = filtersets.DeviceBayFilterSet
561class InventoryItemViewSet(NetBoxModelViewSet):
562 queryset = InventoryItem.objects.all()
563 serializer_class = serializers.InventoryItemSerializer
564 filterset_class = filtersets.InventoryItemFilterSet
567#
568# Device component roles
569#
571class InventoryItemRoleViewSet(NetBoxModelViewSet):
572 queryset = InventoryItemRole.objects.all()
573 serializer_class = serializers.InventoryItemRoleSerializer
574 filterset_class = filtersets.InventoryItemRoleFilterSet
577#
578# Addressing
579#
581class MACAddressViewSet(NetBoxModelViewSet):
582 queryset = MACAddress.objects.all()
583 serializer_class = serializers.MACAddressSerializer
584 filterset_class = filtersets.MACAddressFilterSet
587#
588# Cables
589#
591class CableViewSet(NetBoxModelViewSet):
592 queryset = Cable.objects.prefetch_related('terminations__termination')
593 serializer_class = serializers.CableSerializer
594 filterset_class = filtersets.CableFilterSet
597class CableTerminationViewSet(NetBoxReadOnlyModelViewSet):
598 metadata_class = ContentTypeMetadata
599 queryset = CableTermination.objects.all()
600 serializer_class = serializers.CableTerminationSerializer
601 filterset_class = filtersets.CableTerminationFilterSet
604class CableBundleViewSet(NetBoxModelViewSet):
605 queryset = CableBundle.objects.annotate(
606 cable_count=count_related(Cable, 'bundle')
607 )
608 serializer_class = serializers.CableBundleSerializer
609 filterset_class = filtersets.CableBundleFilterSet
612#
613# Virtual chassis
614#
616class VirtualChassisViewSet(NetBoxModelViewSet):
617 queryset = VirtualChassis.objects.prefetch_related(
618 # Prefetch related object for the display of unnamed devices
619 'master__virtual_chassis',
620 )
621 serializer_class = serializers.VirtualChassisSerializer
622 filterset_class = filtersets.VirtualChassisFilterSet
625#
626# Power panels
627#
629class PowerPanelViewSet(NetBoxModelViewSet):
630 queryset = PowerPanel.objects.all()
631 serializer_class = serializers.PowerPanelSerializer
632 filterset_class = filtersets.PowerPanelFilterSet
635#
636# Power feeds
637#
639class PowerFeedViewSet(PathEndpointMixin, NetBoxModelViewSet):
640 queryset = PowerFeed.objects.prefetch_related(
641 '_path', 'cable__terminations',
642 )
643 serializer_class = serializers.PowerFeedSerializer
644 filterset_class = filtersets.PowerFeedFilterSet
647#
648# Cooling sources
649#
651class CoolingSourceViewSet(NetBoxModelViewSet):
652 queryset = CoolingSource.objects.all()
653 serializer_class = serializers.CoolingSourceSerializer
654 filterset_class = filtersets.CoolingSourceFilterSet
657#
658# Cooling feeds
659#
661class CoolingFeedViewSet(NetBoxModelViewSet):
662 queryset = CoolingFeed.objects.all()
663 serializer_class = serializers.CoolingFeedSerializer
664 filterset_class = filtersets.CoolingFeedFilterSet
667#
668# Miscellaneous
669#
671class ConnectedDeviceViewSet(ViewSet):
672 """
673 This endpoint allows a user to determine what device (if any) is connected to a given peer device and peer
674 interface. This is useful in a situation where a device boots with no configuration, but can detect its neighbors
675 via a protocol such as LLDP. Two query parameters must be included in the request:
677 * `peer_device`: The name of the peer device
678 * `peer_interface`: The name of the peer interface
679 """
680 permission_classes = [IsAuthenticatedOrLoginNotRequired]
681 _device_param = OpenApiParameter(
682 name='peer_device',
683 location='query',
684 description='The name of the peer device',
685 required=True,
686 type=OpenApiTypes.STR
687 )
688 _interface_param = OpenApiParameter(
689 name='peer_interface',
690 location='query',
691 description='The name of the peer interface',
692 required=True,
693 type=OpenApiTypes.STR
694 )
695 serializer_class = serializers.DeviceSerializer
697 def get_view_name(self):
698 return "Connected Device Locator"
700 @extend_schema(
701 parameters=[_device_param, _interface_param],
702 responses={200: serializers.DeviceSerializer}
703 )
704 def list(self, request):
706 peer_device_name = request.query_params.get(self._device_param.name)
707 peer_interface_name = request.query_params.get(self._interface_param.name)
709 if not peer_device_name or not peer_interface_name:
710 raise MissingFilterException(detail='Request must include "peer_device" and "peer_interface" filters.')
712 # Determine local endpoint from peer interface's connection
713 peer_device = get_object_or_404(
714 Device.objects.restrict(request.user, 'view'),
715 name=peer_device_name
716 )
717 peer_interface = get_object_or_404(
718 Interface.objects.restrict(request.user, 'view'),
719 device=peer_device,
720 name=peer_interface_name
721 )
722 endpoints = peer_interface.connected_endpoints
724 # If an Interface, return the parent device
725 if endpoints and type(endpoints[0]) is Interface:
726 device = get_object_or_404(
727 Device.objects.restrict(request.user, 'view'),
728 pk=endpoints[0].device_id
729 )
730 return Response(serializers.DeviceSerializer(device, context={'request': request}).data)
732 # Connected endpoint is none or not an Interface
733 raise Http404