Coverage for ipam/filtersets.py: 48%
625 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import django_filters
2import netaddr
3from django.contrib.contenttypes.models import ContentType
4from django.core.exceptions import ValidationError
5from django.db.models import Q
6from django.utils.translation import gettext as _
7from drf_spectacular.types import OpenApiTypes
8from drf_spectacular.utils import extend_schema_field
9from netaddr.core import AddrFormatError
11from circuits.models import Provider
12from dcim.base_filtersets import ScopedFilterSet
13from dcim.models import Device, Interface, Region, Site, SiteGroup
14from netbox.filtersets import (
15 ChangeLoggedModelFilterSet,
16 NetBoxModelFilterSet,
17 OrganizationalModelFilterSet,
18 PrimaryModelFilterSet,
19)
20from tenancy.filtersets import ContactModelFilterSet, TenancyFilterSet
21from utilities.filters import (
22 MultiValueCharFilter,
23 MultiValueContentTypeFilter,
24 MultiValueNumberFilter,
25 TreeNodeMultipleChoiceFilter,
26)
27from utilities.filtersets import register_filterset
28from virtualization.models import VirtualMachine, VMInterface
29from vpn.models import L2VPN
31from .choices import *
32from .models import *
33from .utils import normalize_port_mapping, port_mapping_q
35__all__ = (
36 'ASNFilterSet',
37 'ASNRangeFilterSet',
38 'AggregateFilterSet',
39 'FHRPGroupAssignmentFilterSet',
40 'FHRPGroupFilterSet',
41 'IPAddressFilterSet',
42 'IPRangeFilterSet',
43 'PrefixFilterSet',
44 'PrimaryIPFilterSet',
45 'RIRFilterSet',
46 'RoleFilterSet',
47 'RouteTargetFilterSet',
48 'ServiceFilterSet',
49 'ServiceTemplateFilterSet',
50 'VLANFilterSet',
51 'VLANGroupFilterSet',
52 'VLANTranslationPolicyFilterSet',
53 'VLANTranslationRuleFilterSet',
54 'VRFFilterSet',
55)
58@register_filterset
59class VRFFilterSet(PrimaryModelFilterSet, TenancyFilterSet):
60 import_target_id = django_filters.ModelMultipleChoiceFilter(
61 field_name='import_targets',
62 queryset=RouteTarget.objects.all(),
63 label=_('Import target'),
64 )
65 import_target = django_filters.ModelMultipleChoiceFilter(
66 field_name='import_targets__name',
67 queryset=RouteTarget.objects.all(),
68 to_field_name='name',
69 label=_('Import target (name)'),
70 )
71 export_target_id = django_filters.ModelMultipleChoiceFilter(
72 field_name='export_targets',
73 queryset=RouteTarget.objects.all(),
74 label=_('Export target'),
75 )
76 export_target = django_filters.ModelMultipleChoiceFilter(
77 field_name='export_targets__name',
78 queryset=RouteTarget.objects.all(),
79 to_field_name='name',
80 label=_('Export target (name)'),
81 )
83 def search(self, queryset, name, value):
84 if not value.strip():
85 return queryset
86 return queryset.filter(
87 Q(name__icontains=value) |
88 Q(rd__icontains=value) |
89 Q(description__icontains=value)
90 )
92 class Meta:
93 model = VRF
94 fields = ('id', 'name', 'rd', 'enforce_unique', 'description')
97@register_filterset
98class RouteTargetFilterSet(PrimaryModelFilterSet, TenancyFilterSet):
99 importing_vrf_id = django_filters.ModelMultipleChoiceFilter(
100 field_name='importing_vrfs',
101 queryset=VRF.objects.all(),
102 label=_('Importing VRF'),
103 )
104 importing_vrf = django_filters.ModelMultipleChoiceFilter(
105 field_name='importing_vrfs__rd',
106 queryset=VRF.objects.all(),
107 to_field_name='rd',
108 label=_('Import VRF (RD)'),
109 )
110 exporting_vrf_id = django_filters.ModelMultipleChoiceFilter(
111 field_name='exporting_vrfs',
112 queryset=VRF.objects.all(),
113 label=_('Exporting VRF'),
114 )
115 exporting_vrf = django_filters.ModelMultipleChoiceFilter(
116 field_name='exporting_vrfs__rd',
117 queryset=VRF.objects.all(),
118 to_field_name='rd',
119 label=_('Export VRF (RD)'),
120 )
121 importing_l2vpn_id = django_filters.ModelMultipleChoiceFilter(
122 field_name='importing_l2vpns',
123 queryset=L2VPN.objects.all(),
124 label=_('Importing L2VPN'),
125 )
126 importing_l2vpn = django_filters.ModelMultipleChoiceFilter(
127 field_name='importing_l2vpns__identifier',
128 queryset=L2VPN.objects.all(),
129 to_field_name='identifier',
130 label=_('Importing L2VPN (identifier)'),
131 )
132 exporting_l2vpn_id = django_filters.ModelMultipleChoiceFilter(
133 field_name='exporting_l2vpns',
134 queryset=L2VPN.objects.all(),
135 label=_('Exporting L2VPN'),
136 )
137 exporting_l2vpn = django_filters.ModelMultipleChoiceFilter(
138 field_name='exporting_l2vpns__identifier',
139 queryset=L2VPN.objects.all(),
140 to_field_name='identifier',
141 label=_('Exporting L2VPN (identifier)'),
142 )
144 def search(self, queryset, name, value):
145 if not value.strip():
146 return queryset
147 return queryset.filter(
148 Q(name__icontains=value) |
149 Q(description__icontains=value)
150 )
152 class Meta:
153 model = RouteTarget
154 fields = ('id', 'name', 'description')
157@register_filterset
158class RIRFilterSet(OrganizationalModelFilterSet):
160 class Meta:
161 model = RIR
162 fields = ('id', 'name', 'slug', 'is_private', 'description')
165@register_filterset
166class AggregateFilterSet(PrimaryModelFilterSet, TenancyFilterSet, ContactModelFilterSet):
167 family = django_filters.NumberFilter(
168 field_name='prefix',
169 lookup_expr='family'
170 )
171 prefix = django_filters.CharFilter(
172 method='filter_prefix',
173 label=_('Prefix'),
174 )
175 rir_id = django_filters.ModelMultipleChoiceFilter(
176 queryset=RIR.objects.all(),
177 distinct=False,
178 label=_('RIR (ID)'),
179 )
180 rir = django_filters.ModelMultipleChoiceFilter(
181 field_name='rir__slug',
182 queryset=RIR.objects.all(),
183 distinct=False,
184 to_field_name='slug',
185 label=_('RIR (slug)'),
186 )
188 class Meta:
189 model = Aggregate
190 fields = ('id', 'date_added', 'description')
192 def search(self, queryset, name, value):
193 if not value.strip():
194 return queryset
195 qs_filter = Q(description__icontains=value)
196 qs_filter |= Q(prefix__contains=value.strip())
197 try:
198 prefix = str(netaddr.IPNetwork(value.strip()).cidr)
199 qs_filter |= Q(prefix__net_contains_or_equals=prefix)
200 qs_filter |= Q(prefix__contains=value.strip())
201 except (AddrFormatError, ValueError):
202 pass
203 return queryset.filter(qs_filter)
205 def filter_prefix(self, queryset, name, value):
206 if not value.strip():
207 return queryset
208 try:
209 query = str(netaddr.IPNetwork(value).cidr)
210 return queryset.filter(prefix=query)
211 except (AddrFormatError, ValueError):
212 return queryset.none()
215@register_filterset
216class ASNRangeFilterSet(OrganizationalModelFilterSet, TenancyFilterSet):
217 rir_id = django_filters.ModelMultipleChoiceFilter(
218 queryset=RIR.objects.all(),
219 distinct=False,
220 label=_('RIR (ID)'),
221 )
222 rir = django_filters.ModelMultipleChoiceFilter(
223 field_name='rir__slug',
224 queryset=RIR.objects.all(),
225 distinct=False,
226 to_field_name='slug',
227 label=_('RIR (slug)'),
228 )
230 class Meta:
231 model = ASNRange
232 fields = ('id', 'name', 'slug', 'start', 'end', 'description')
234 def search(self, queryset, name, value):
235 if not value.strip():
236 return queryset
237 return queryset.filter(
238 Q(name__icontains=value) |
239 Q(description__icontains=value)
240 )
243@register_filterset
244class ASNFilterSet(PrimaryModelFilterSet, TenancyFilterSet):
245 rir_id = django_filters.ModelMultipleChoiceFilter(
246 queryset=RIR.objects.all(),
247 distinct=False,
248 label=_('RIR (ID)'),
249 )
250 rir = django_filters.ModelMultipleChoiceFilter(
251 field_name='rir__slug',
252 queryset=RIR.objects.all(),
253 distinct=False,
254 to_field_name='slug',
255 label=_('RIR (slug)'),
256 )
257 site_group_id = TreeNodeMultipleChoiceFilter(
258 queryset=SiteGroup.objects.all(),
259 field_name='sites__group',
260 lookup_expr='in',
261 label=_('Site group (ID)'),
262 )
263 site_group = TreeNodeMultipleChoiceFilter(
264 queryset=SiteGroup.objects.all(),
265 field_name='sites__group',
266 lookup_expr='in',
267 to_field_name='slug',
268 label=_('Site group (slug)'),
269 )
270 site_id = django_filters.ModelMultipleChoiceFilter(
271 field_name='sites',
272 queryset=Site.objects.all(),
273 label=_('Site (ID)'),
274 )
275 site = django_filters.ModelMultipleChoiceFilter(
276 field_name='sites__slug',
277 queryset=Site.objects.all(),
278 to_field_name='slug',
279 label=_('Site (slug)'),
280 )
281 provider_id = django_filters.ModelMultipleChoiceFilter(
282 field_name='providers',
283 queryset=Provider.objects.all(),
284 label=_('Provider (ID)'),
285 )
286 provider = django_filters.ModelMultipleChoiceFilter(
287 field_name='providers__slug',
288 queryset=Provider.objects.all(),
289 to_field_name='slug',
290 label=_('Provider (slug)'),
291 )
292 role_id = django_filters.ModelMultipleChoiceFilter(
293 queryset=Role.objects.all(),
294 distinct=False,
295 label=_('Role (ID)'),
296 )
297 role = django_filters.ModelMultipleChoiceFilter(
298 field_name='role__slug',
299 queryset=Role.objects.all(),
300 distinct=False,
301 to_field_name='slug',
302 label=_('Role (slug)'),
303 )
305 class Meta:
306 model = ASN
307 fields = ('id', 'asn', 'description')
309 def search(self, queryset, name, value):
310 if not value.strip():
311 return queryset
312 qs_filter = Q(description__icontains=value)
313 try:
314 qs_filter |= Q(asn=int(value))
315 except ValueError:
316 pass
317 return queryset.filter(qs_filter)
320@register_filterset
321class RoleFilterSet(OrganizationalModelFilterSet):
323 class Meta:
324 model = Role
325 fields = ('id', 'name', 'slug', 'description', 'weight')
328@register_filterset
329class PrefixFilterSet(PrimaryModelFilterSet, ScopedFilterSet, TenancyFilterSet, ContactModelFilterSet):
330 family = django_filters.NumberFilter(
331 field_name='prefix',
332 lookup_expr='family'
333 )
334 prefix = MultiValueCharFilter(
335 method='filter_prefix',
336 label=_('Prefix'),
337 )
338 within = django_filters.CharFilter(
339 method='search_within',
340 label=_('Within prefix'),
341 )
342 within_include = django_filters.CharFilter(
343 method='search_within_include',
344 label=_('Within and including prefix'),
345 )
346 contains = django_filters.CharFilter(
347 method='search_contains',
348 label=_('Prefixes which contain this prefix or IP'),
349 )
350 depth = MultiValueNumberFilter(
351 field_name='_depth'
352 )
353 children = MultiValueNumberFilter(
354 field_name='_children'
355 )
356 mask_length = MultiValueNumberFilter(
357 field_name='prefix',
358 lookup_expr='net_mask_length',
359 label=_('Mask length')
360 )
361 mask_length__gte = django_filters.NumberFilter(
362 field_name='prefix',
363 lookup_expr='net_mask_length__gte'
364 )
365 mask_length__lte = django_filters.NumberFilter(
366 field_name='prefix',
367 lookup_expr='net_mask_length__lte'
368 )
369 vrf_id = django_filters.ModelMultipleChoiceFilter(
370 queryset=VRF.objects.all(),
371 distinct=False,
372 label=_('VRF'),
373 )
374 vrf = django_filters.ModelMultipleChoiceFilter(
375 field_name='vrf__rd',
376 queryset=VRF.objects.all(),
377 distinct=False,
378 to_field_name='rd',
379 label=_('VRF (RD)'),
380 )
381 present_in_vrf_id = django_filters.ModelChoiceFilter(
382 queryset=VRF.objects.all(),
383 method='filter_present_in_vrf',
384 label=_('VRF')
385 )
386 present_in_vrf = django_filters.ModelChoiceFilter(
387 queryset=VRF.objects.all(),
388 method='filter_present_in_vrf',
389 to_field_name='rd',
390 label=_('VRF (RD)'),
391 )
392 vlan_group_id = django_filters.ModelMultipleChoiceFilter(
393 field_name='vlan__group',
394 queryset=VLANGroup.objects.all(),
395 distinct=False,
396 to_field_name='id',
397 label=_('VLAN Group (ID)'),
398 )
399 vlan_group = django_filters.ModelMultipleChoiceFilter(
400 field_name='vlan__group__slug',
401 queryset=VLANGroup.objects.all(),
402 distinct=False,
403 to_field_name='slug',
404 label=_('VLAN Group (slug)'),
405 )
406 vlan_id = django_filters.ModelMultipleChoiceFilter(
407 queryset=VLAN.objects.all(),
408 distinct=False,
409 label=_('VLAN (ID)'),
410 )
411 vlan_vid = django_filters.NumberFilter(
412 field_name='vlan__vid',
413 label=_('VLAN number (1-4094)'),
414 )
415 role_id = django_filters.ModelMultipleChoiceFilter(
416 queryset=Role.objects.all(),
417 distinct=False,
418 label=_('Role (ID)'),
419 )
420 role = django_filters.ModelMultipleChoiceFilter(
421 field_name='role__slug',
422 queryset=Role.objects.all(),
423 distinct=False,
424 to_field_name='slug',
425 label=_('Role (slug)'),
426 )
427 status = django_filters.MultipleChoiceFilter(
428 choices=PrefixStatusChoices,
429 distinct=False,
430 null_value=None
431 )
433 class Meta:
434 model = Prefix
435 fields = ('id', 'scope_id', 'is_pool', 'mark_utilized', 'description')
437 def search(self, queryset, name, value):
438 if not value.strip():
439 return queryset
440 qs_filter = Q(description__icontains=value)
441 qs_filter |= Q(prefix__contains=value.strip())
442 try:
443 prefix = str(netaddr.IPNetwork(value.strip()).cidr)
444 qs_filter |= Q(prefix__net_contains_or_equals=prefix)
445 qs_filter |= Q(prefix__contains=value.strip())
446 except (AddrFormatError, ValueError):
447 pass
448 return queryset.filter(qs_filter)
450 def filter_prefix(self, queryset, name, value):
451 query_values = []
452 for v in value:
453 try:
454 query_values.append(netaddr.IPNetwork(v))
455 except (AddrFormatError, ValueError):
456 pass
457 return queryset.filter(prefix__in=query_values)
459 def search_within(self, queryset, name, value):
460 value = value.strip()
461 if not value:
462 return queryset
463 try:
464 query = str(netaddr.IPNetwork(value).cidr)
465 return queryset.filter(prefix__net_contained=query)
466 except (AddrFormatError, ValueError):
467 return queryset.none()
469 def search_within_include(self, queryset, name, value):
470 value = value.strip()
471 if not value:
472 return queryset
473 try:
474 query = str(netaddr.IPNetwork(value).cidr)
475 return queryset.filter(prefix__net_contained_or_equal=query)
476 except (AddrFormatError, ValueError):
477 return queryset.none()
479 def search_contains(self, queryset, name, value):
480 value = value.strip()
481 if not value:
482 return queryset
483 try:
484 # Searching by prefix
485 if '/' in value:
486 return queryset.filter(prefix__net_contains_or_equals=str(netaddr.IPNetwork(value).cidr))
487 # Searching by IP address
488 return queryset.filter(prefix__net_contains=str(netaddr.IPAddress(value)))
489 except (AddrFormatError, ValueError):
490 return queryset.none()
492 @extend_schema_field(OpenApiTypes.STR)
493 def filter_present_in_vrf(self, queryset, name, vrf):
494 if vrf is None:
495 return queryset.none()
496 return queryset.filter(
497 Q(vrf=vrf) |
498 Q(vrf__export_targets__in=vrf.import_targets.all())
499 ).distinct()
502@register_filterset
503class IPRangeFilterSet(PrimaryModelFilterSet, TenancyFilterSet, ContactModelFilterSet):
504 family = django_filters.NumberFilter(
505 field_name='start_address',
506 lookup_expr='family'
507 )
508 start_address = MultiValueCharFilter(
509 method='filter_address',
510 label=_('Address'),
511 )
512 end_address = MultiValueCharFilter(
513 method='filter_address',
514 label=_('Address'),
515 )
516 contains = django_filters.CharFilter(
517 method='search_contains',
518 label=_('Ranges which contain this prefix or IP'),
519 )
520 vrf_id = django_filters.ModelMultipleChoiceFilter(
521 queryset=VRF.objects.all(),
522 distinct=False,
523 label=_('VRF'),
524 )
525 vrf = django_filters.ModelMultipleChoiceFilter(
526 field_name='vrf__rd',
527 queryset=VRF.objects.all(),
528 distinct=False,
529 to_field_name='rd',
530 label=_('VRF (RD)'),
531 )
532 role_id = django_filters.ModelMultipleChoiceFilter(
533 queryset=Role.objects.all(),
534 distinct=False,
535 label=_('Role (ID)'),
536 )
537 role = django_filters.ModelMultipleChoiceFilter(
538 field_name='role__slug',
539 queryset=Role.objects.all(),
540 distinct=False,
541 to_field_name='slug',
542 label=_('Role (slug)'),
543 )
544 status = django_filters.MultipleChoiceFilter(
545 choices=IPRangeStatusChoices,
546 distinct=False,
547 null_value=None
548 )
549 parent = MultiValueCharFilter(
550 method='search_by_parent',
551 label=_('Parent prefix'),
552 )
554 class Meta:
555 model = IPRange
556 fields = ('id', 'mark_populated', 'mark_utilized', 'size', 'description')
558 def search(self, queryset, name, value):
559 if not value.strip():
560 return queryset
561 qs_filter = Q(description__icontains=value) | Q(start_address__contains=value) | Q(end_address__contains=value)
562 try:
563 ipaddress = str(netaddr.IPNetwork(value.strip()))
564 qs_filter |= Q(start_address=ipaddress)
565 qs_filter |= Q(end_address=ipaddress)
566 except (AddrFormatError, ValueError):
567 pass
568 return queryset.filter(qs_filter)
570 def search_contains(self, queryset, name, value):
571 value = value.strip()
572 if not value:
573 return queryset
574 try:
575 # Strip mask
576 ipaddress = netaddr.IPNetwork(value)
577 return queryset.filter(start_address__lte=ipaddress, end_address__gte=ipaddress)
578 except (AddrFormatError, ValueError):
579 return queryset.none()
581 def filter_address(self, queryset, name, value):
582 try:
583 return queryset.filter(**{f'{name}__net_in': value})
584 except ValidationError:
585 return queryset.none()
587 def search_by_parent(self, queryset, name, value):
588 if not value:
589 return queryset
590 q = Q()
591 for prefix in value:
592 try:
593 query = str(netaddr.IPNetwork(prefix.strip()).cidr)
594 q |= Q(start_address__net_host_contained=query, end_address__net_host_contained=query)
595 except (AddrFormatError, ValueError):
596 return queryset.none()
597 return queryset.filter(q)
600@register_filterset
601class IPAddressFilterSet(PrimaryModelFilterSet, TenancyFilterSet, ContactModelFilterSet):
602 family = django_filters.NumberFilter(
603 field_name='address',
604 lookup_expr='family'
605 )
606 parent = MultiValueCharFilter(
607 method='search_by_parent',
608 label=_('Parent prefix'),
609 )
610 address = MultiValueCharFilter(
611 method='filter_address',
612 label=_('Address'),
613 )
614 mask_length = MultiValueNumberFilter(
615 field_name='address',
616 lookup_expr='net_mask_length',
617 label=_('Mask length')
618 )
619 mask_length__gte = django_filters.NumberFilter(
620 field_name='address',
621 lookup_expr='net_mask_length__gte'
622 )
623 mask_length__lte = django_filters.NumberFilter(
624 field_name='address',
625 lookup_expr='net_mask_length__lte'
626 )
627 vrf_id = django_filters.ModelMultipleChoiceFilter(
628 queryset=VRF.objects.all(),
629 distinct=False,
630 label=_('VRF'),
631 )
632 vrf = django_filters.ModelMultipleChoiceFilter(
633 field_name='vrf__rd',
634 queryset=VRF.objects.all(),
635 distinct=False,
636 to_field_name='rd',
637 label=_('VRF (RD)'),
638 )
639 present_in_vrf_id = django_filters.ModelChoiceFilter(
640 queryset=VRF.objects.all(),
641 method='filter_present_in_vrf',
642 label=_('VRF')
643 )
644 present_in_vrf = django_filters.ModelChoiceFilter(
645 queryset=VRF.objects.all(),
646 method='filter_present_in_vrf',
647 to_field_name='rd',
648 label=_('VRF (RD)'),
649 )
650 assigned_object_type = MultiValueContentTypeFilter()
651 device = MultiValueCharFilter(
652 method='filter_device',
653 field_name='name',
654 label=_('Device (name)'),
655 )
656 device_id = MultiValueNumberFilter(
657 method='filter_device',
658 field_name='pk',
659 label=_('Device (ID)'),
660 )
661 virtual_machine = MultiValueCharFilter(
662 method='filter_virtual_machine',
663 field_name='name',
664 label=_('Virtual machine (name)'),
665 )
666 virtual_machine_id = MultiValueNumberFilter(
667 method='filter_virtual_machine',
668 field_name='pk',
669 label=_('Virtual machine (ID)'),
670 )
671 interface = django_filters.ModelMultipleChoiceFilter(
672 field_name='interface__name',
673 queryset=Interface.objects.all(),
674 to_field_name='name',
675 label=_('Interface (name)'),
676 )
677 interface_id = django_filters.ModelMultipleChoiceFilter(
678 field_name='interface',
679 queryset=Interface.objects.all(),
680 label=_('Interface (ID)'),
681 )
682 vminterface = django_filters.ModelMultipleChoiceFilter(
683 field_name='vminterface__name',
684 queryset=VMInterface.objects.all(),
685 to_field_name='name',
686 label=_('VM interface (name)'),
687 )
688 vminterface_id = django_filters.ModelMultipleChoiceFilter(
689 field_name='vminterface',
690 queryset=VMInterface.objects.all(),
691 label=_('VM interface (ID)'),
692 )
693 fhrpgroup_id = django_filters.ModelMultipleChoiceFilter(
694 field_name='fhrpgroup',
695 queryset=FHRPGroup.objects.all(),
696 label=_('FHRP group (ID)'),
697 )
698 assigned_to_interface = django_filters.BooleanFilter(
699 method='_assigned_to_interface',
700 label=_('Is assigned to an interface'),
701 )
702 assigned = django_filters.BooleanFilter(
703 method='_assigned',
704 label=_('Is assigned'),
705 )
706 status = django_filters.MultipleChoiceFilter(
707 choices=IPAddressStatusChoices,
708 distinct=False,
709 null_value=None
710 )
711 role = django_filters.MultipleChoiceFilter(
712 choices=IPAddressRoleChoices,
713 distinct=False,
714 )
715 service_id = django_filters.ModelMultipleChoiceFilter(
716 field_name='services',
717 queryset=Service.objects.all(),
718 label=_('Application Service (ID)'),
719 )
720 nat_inside_id = django_filters.ModelMultipleChoiceFilter(
721 field_name='nat_inside',
722 queryset=IPAddress.objects.all(),
723 distinct=False,
724 label=_('NAT inside IP address (ID)'),
725 )
727 class Meta:
728 model = IPAddress
729 fields = ('id', 'dns_name', 'description', 'assigned_object_type', 'assigned_object_id')
731 def search(self, queryset, name, value):
732 if not value.strip():
733 return queryset
734 qs_filter = (
735 Q(dns_name__icontains=value) |
736 Q(description__icontains=value) |
737 Q(address__istartswith=value)
738 )
739 return queryset.filter(qs_filter)
741 def search_by_parent(self, queryset, name, value):
742 if not value:
743 return queryset
744 q = Q()
745 for prefix in value:
746 try:
747 query = str(netaddr.IPNetwork(prefix.strip()).cidr)
748 q |= Q(address__net_host_contained=query)
749 except (AddrFormatError, ValueError):
750 return queryset.none()
751 return queryset.filter(q)
753 def parse_inet_addresses(self, value):
754 """
755 Parse networks or IP addresses and cast to a format
756 acceptable by the Postgres inet type.
758 Skips invalid values.
759 """
760 parsed = []
761 for addr in value:
762 if netaddr.valid_ipv4(addr) or netaddr.valid_ipv6(addr):
763 parsed.append(addr)
764 continue
765 try:
766 network = netaddr.IPNetwork(addr)
767 parsed.append(str(network))
768 except (AddrFormatError, ValueError):
769 continue
770 return parsed
772 def filter_address(self, queryset, name, value):
773 # Let's first parse the addresses passed
774 # as argument. If they are all invalid,
775 # we return an empty queryset
776 value = self.parse_inet_addresses(value)
777 if len(value) == 0:
778 return queryset.none()
780 try:
781 return queryset.filter(address__net_in=value)
782 except ValidationError:
783 return queryset.none()
785 @extend_schema_field(OpenApiTypes.STR)
786 def filter_present_in_vrf(self, queryset, name, vrf):
787 if vrf is None:
788 return queryset.none()
789 return queryset.filter(
790 Q(vrf=vrf) |
791 Q(vrf__export_targets__in=vrf.import_targets.all())
792 ).distinct()
794 def filter_device(self, queryset, name, value):
795 devices = Device.objects.filter(**{'{}__in'.format(name): value})
796 if not devices.exists():
797 return queryset.none()
798 interface_ids = []
799 for device in devices:
800 interface_ids.extend(device.vc_interfaces().values_list('id', flat=True))
801 return queryset.filter(
802 interface__in=interface_ids
803 )
805 def filter_virtual_machine(self, queryset, name, value):
806 virtual_machines = VirtualMachine.objects.filter(**{'{}__in'.format(name): value})
807 if not virtual_machines.exists():
808 return queryset.none()
809 interface_ids = []
810 for vm in virtual_machines:
811 interface_ids.extend(vm.interfaces.values_list('id', flat=True))
812 return queryset.filter(
813 vminterface__in=interface_ids
814 )
816 def _assigned_to_interface(self, queryset, name, value):
817 content_types = ContentType.objects.get_for_models(Interface, VMInterface).values()
818 if value:
819 return queryset.filter(
820 assigned_object_type__in=content_types,
821 assigned_object_id__isnull=False
822 )
823 return queryset.exclude(
824 assigned_object_type__in=content_types,
825 assigned_object_id__isnull=False
826 )
828 def _assigned(self, queryset, name, value):
829 if value:
830 return queryset.exclude(
831 assigned_object_type__isnull=True,
832 assigned_object_id__isnull=True
833 )
834 return queryset.filter(
835 assigned_object_type__isnull=True,
836 assigned_object_id__isnull=True
837 )
840@register_filterset
841class FHRPGroupFilterSet(PrimaryModelFilterSet):
842 protocol = django_filters.MultipleChoiceFilter(
843 choices=FHRPGroupProtocolChoices,
844 distinct=False,
845 )
846 auth_type = django_filters.MultipleChoiceFilter(
847 choices=FHRPGroupAuthTypeChoices,
848 distinct=False,
849 )
850 related_ip = django_filters.ModelMultipleChoiceFilter(
851 queryset=IPAddress.objects.all(),
852 method='filter_related_ip'
853 )
855 class Meta:
856 model = FHRPGroup
857 fields = ('id', 'group_id', 'name', 'auth_key', 'description')
859 def search(self, queryset, name, value):
860 if not value.strip():
861 return queryset
862 return queryset.filter(
863 Q(description__icontains=value) |
864 Q(group_id__contains=value) |
865 Q(name__icontains=value)
866 )
868 @extend_schema_field(OpenApiTypes.STR)
869 def filter_related_ip(self, queryset, name, value):
870 """
871 Filter by VRF & prefix of assigned IP addresses.
872 """
873 ip_filter = Q()
874 for ipaddress in value: 874 ↛ 875line 874 didn't jump to line 875 because the loop on line 874 never started
875 if ipaddress.vrf:
876 q = Q(
877 ip_addresses__address__net_contained_or_equal=ipaddress.address,
878 ip_addresses__vrf=ipaddress.vrf
879 )
880 else:
881 q = Q(
882 ip_addresses__address__net_contained_or_equal=ipaddress.address,
883 ip_addresses__vrf__isnull=True
884 )
885 ip_filter |= q
887 return queryset.filter(ip_filter)
890@register_filterset
891class FHRPGroupAssignmentFilterSet(ChangeLoggedModelFilterSet):
892 interface_type = MultiValueContentTypeFilter()
893 group_id = django_filters.ModelMultipleChoiceFilter(
894 queryset=FHRPGroup.objects.all(),
895 distinct=False,
896 label=_('Group (ID)'),
897 )
898 device = MultiValueCharFilter(
899 method='filter_device',
900 field_name='name',
901 label=_('Device (name)'),
902 )
903 device_id = MultiValueNumberFilter(
904 method='filter_device',
905 field_name='pk',
906 label=_('Device (ID)'),
907 )
908 virtual_machine = MultiValueCharFilter(
909 method='filter_virtual_machine',
910 field_name='name',
911 label=_('Virtual machine (name)'),
912 )
913 virtual_machine_id = MultiValueNumberFilter(
914 method='filter_virtual_machine',
915 field_name='pk',
916 label=_('Virtual machine (ID)'),
917 )
919 class Meta:
920 model = FHRPGroupAssignment
921 fields = ('id', 'group_id', 'interface_type', 'interface_id', 'priority')
923 def filter_device(self, queryset, name, value):
924 devices = Device.objects.filter(**{f'{name}__in': value})
925 if not devices.exists():
926 return queryset.none()
927 interface_ids = []
928 for device in devices:
929 interface_ids.extend(device.vc_interfaces().values_list('id', flat=True))
930 return queryset.filter(
931 Q(interface_type=ContentType.objects.get_for_model(Interface), interface_id__in=interface_ids)
932 )
934 def filter_virtual_machine(self, queryset, name, value):
935 virtual_machines = VirtualMachine.objects.filter(**{f'{name}__in': value})
936 if not virtual_machines.exists():
937 return queryset.none()
938 interface_ids = []
939 for vm in virtual_machines:
940 interface_ids.extend(vm.interfaces.values_list('id', flat=True))
941 return queryset.filter(
942 Q(interface_type=ContentType.objects.get_for_model(VMInterface), interface_id__in=interface_ids)
943 )
946@register_filterset
947class VLANGroupFilterSet(OrganizationalModelFilterSet, TenancyFilterSet):
948 scope_type = MultiValueContentTypeFilter()
949 region = MultiValueNumberFilter(
950 method='filter_scope'
951 )
952 site_group = MultiValueNumberFilter(
953 method='filter_scope'
954 )
955 site = MultiValueNumberFilter(
956 method='filter_scope'
957 )
958 location = MultiValueNumberFilter(
959 method='filter_scope'
960 )
961 rack_group = MultiValueNumberFilter(
962 method='filter_scope'
963 )
964 rack = MultiValueNumberFilter(
965 method='filter_scope'
966 )
967 cluster_group = MultiValueNumberFilter(
968 method='filter_scope'
969 )
970 cluster = MultiValueNumberFilter(
971 method='filter_scope'
972 )
973 contains_vid = django_filters.NumberFilter(
974 field_name='vid_ranges',
975 lookup_expr='range_contains',
976 )
978 class Meta:
979 model = VLANGroup
980 fields = ('id', 'name', 'slug', 'description', 'scope_id', 'total_vlan_ids')
982 def search(self, queryset, name, value):
983 if not value.strip():
984 return queryset
985 qs_filter = (
986 Q(name__icontains=value) |
987 Q(description__icontains=value)
988 )
989 return queryset.filter(qs_filter)
991 def filter_scope(self, queryset, name, value):
992 model_name = name.replace('_', '')
993 return queryset.filter(
994 scope_type=ContentType.objects.get(model=model_name),
995 scope_id__in=value
996 )
999@register_filterset
1000class VLANFilterSet(PrimaryModelFilterSet, TenancyFilterSet):
1001 region_id = TreeNodeMultipleChoiceFilter(
1002 queryset=Region.objects.all(),
1003 field_name='site__region',
1004 lookup_expr='in',
1005 label=_('Region (ID)'),
1006 )
1007 region = TreeNodeMultipleChoiceFilter(
1008 queryset=Region.objects.all(),
1009 field_name='site__region',
1010 lookup_expr='in',
1011 to_field_name='slug',
1012 label=_('Region (slug)'),
1013 )
1014 site_group_id = TreeNodeMultipleChoiceFilter(
1015 queryset=SiteGroup.objects.all(),
1016 field_name='site__group',
1017 lookup_expr='in',
1018 label=_('Site group (ID)'),
1019 )
1020 site_group = TreeNodeMultipleChoiceFilter(
1021 queryset=SiteGroup.objects.all(),
1022 field_name='site__group',
1023 lookup_expr='in',
1024 to_field_name='slug',
1025 label=_('Site group (slug)'),
1026 )
1027 site_id = django_filters.ModelMultipleChoiceFilter(
1028 queryset=Site.objects.all(),
1029 distinct=False,
1030 label=_('Site (ID)'),
1031 )
1032 site = django_filters.ModelMultipleChoiceFilter(
1033 field_name='site__slug',
1034 queryset=Site.objects.all(),
1035 distinct=False,
1036 to_field_name='slug',
1037 label=_('Site (slug)'),
1038 )
1039 related_to_site = django_filters.ModelMultipleChoiceFilter(
1040 queryset=Site.objects.all(),
1041 method='filter_related_to_site',
1042 label=_('Related to site (ID)'),
1043 )
1044 # get_additional_lookups() skips filters with a method.
1045 related_to_site__n = django_filters.ModelMultipleChoiceFilter(
1046 queryset=Site.objects.all(),
1047 method='filter_related_to_site_negated',
1048 label=_('Related to site (ID)'),
1049 )
1050 group_id = django_filters.ModelMultipleChoiceFilter(
1051 queryset=VLANGroup.objects.all(),
1052 distinct=False,
1053 label=_('Group (ID)'),
1054 )
1055 group = django_filters.ModelMultipleChoiceFilter(
1056 field_name='group__slug',
1057 queryset=VLANGroup.objects.all(),
1058 distinct=False,
1059 to_field_name='slug',
1060 label=_('Group'),
1061 )
1062 role_id = django_filters.ModelMultipleChoiceFilter(
1063 queryset=Role.objects.all(),
1064 distinct=False,
1065 label=_('Role (ID)'),
1066 )
1067 role = django_filters.ModelMultipleChoiceFilter(
1068 field_name='role__slug',
1069 queryset=Role.objects.all(),
1070 distinct=False,
1071 to_field_name='slug',
1072 label=_('Role (slug)'),
1073 )
1074 status = django_filters.MultipleChoiceFilter(
1075 choices=VLANStatusChoices,
1076 distinct=False,
1077 null_value=None
1078 )
1079 available_at_site = django_filters.ModelChoiceFilter(
1080 queryset=Site.objects.all(),
1081 method='get_for_site'
1082 )
1083 available_at_site_group = django_filters.ModelChoiceFilter(
1084 queryset=SiteGroup.objects.all(),
1085 method='get_for_site_group'
1086 )
1087 available_on_device = django_filters.ModelChoiceFilter(
1088 queryset=Device.objects.select_related('cluster'),
1089 method='get_for_device'
1090 )
1091 available_on_virtualmachine = django_filters.ModelChoiceFilter(
1092 queryset=VirtualMachine.objects.all(),
1093 method='get_for_virtualmachine'
1094 )
1095 qinq_role = django_filters.MultipleChoiceFilter(
1096 choices=VLANQinQRoleChoices,
1097 distinct=False,
1098 )
1099 qinq_svlan_id = django_filters.ModelMultipleChoiceFilter(
1100 queryset=VLAN.objects.all(),
1101 distinct=False,
1102 label=_('Q-in-Q SVLAN (ID)'),
1103 )
1104 qinq_svlan_vid = MultiValueNumberFilter(
1105 field_name='qinq_svlan__vid',
1106 label=_('Q-in-Q SVLAN number (1-4094)'),
1107 )
1108 l2vpn_id = django_filters.ModelMultipleChoiceFilter(
1109 field_name='l2vpn_terminations__l2vpn',
1110 queryset=L2VPN.objects.all(),
1111 label=_('L2VPN (ID)'),
1112 )
1113 l2vpn = django_filters.ModelMultipleChoiceFilter(
1114 field_name='l2vpn_terminations__l2vpn__identifier',
1115 queryset=L2VPN.objects.all(),
1116 to_field_name='identifier',
1117 label=_('L2VPN'),
1118 )
1119 interface_id = django_filters.ModelChoiceFilter(
1120 queryset=Interface.objects.all(),
1121 method='filter_interface_id',
1122 label=_('Assigned interface')
1123 )
1124 vminterface_id = django_filters.ModelChoiceFilter(
1125 queryset=VMInterface.objects.all(),
1126 method='filter_vminterface_id',
1127 label=_('Assigned VM interface')
1128 )
1130 class Meta:
1131 model = VLAN
1132 fields = ('id', 'vid', 'name', 'description')
1134 def search(self, queryset, name, value):
1135 if not value.strip():
1136 return queryset
1137 qs_filter = Q(name__icontains=value) | Q(description__icontains=value)
1138 try:
1139 qs_filter |= Q(vid=int(value.strip()))
1140 except ValueError:
1141 pass
1142 return queryset.filter(qs_filter)
1144 @extend_schema_field(OpenApiTypes.STR)
1145 def get_for_site(self, queryset, name, value):
1146 return queryset.get_for_site(value)
1148 @extend_schema_field(OpenApiTypes.STR)
1149 def get_for_site_group(self, queryset, name, value):
1150 return queryset.get_for_site_group(value)
1152 @extend_schema_field(OpenApiTypes.STR)
1153 def get_for_device(self, queryset, name, value):
1154 return queryset.get_for_device(value)
1156 @extend_schema_field(OpenApiTypes.STR)
1157 def get_for_virtualmachine(self, queryset, name, value):
1158 return queryset.get_for_virtualmachine(value)
1160 @extend_schema_field(OpenApiTypes.INT)
1161 def filter_related_to_site(self, queryset, name, value):
1162 if not value: 1162 ↛ 1164line 1162 didn't jump to line 1164 because the condition on line 1162 was always true
1163 return queryset
1164 return queryset.get_related_to_sites(value)
1166 @extend_schema_field(OpenApiTypes.INT)
1167 def filter_related_to_site_negated(self, queryset, name, value):
1168 if not value: 1168 ↛ 1170line 1168 didn't jump to line 1170 because the condition on line 1168 was always true
1169 return queryset
1170 return queryset.get_related_to_sites(value, negate=True)
1172 @extend_schema_field(OpenApiTypes.INT)
1173 def filter_interface_id(self, queryset, name, value):
1174 if value is None:
1175 return queryset.none()
1176 tagged = queryset.filter(interfaces_as_tagged=value)
1177 untagged = queryset.filter(interfaces_as_untagged=value)
1178 return queryset.filter(pk__in=tagged.union(untagged).values('pk'))
1180 @extend_schema_field(OpenApiTypes.INT)
1181 def filter_vminterface_id(self, queryset, name, value):
1182 if value is None:
1183 return queryset.none()
1184 tagged = queryset.filter(vminterfaces_as_tagged=value)
1185 untagged = queryset.filter(vminterfaces_as_untagged=value)
1186 return queryset.filter(pk__in=tagged.union(untagged).values('pk'))
1189@register_filterset
1190class VLANTranslationPolicyFilterSet(PrimaryModelFilterSet):
1192 class Meta:
1193 model = VLANTranslationPolicy
1194 fields = ('id', 'name', 'description')
1196 def search(self, queryset, name, value):
1197 if not value.strip():
1198 return queryset
1199 qs_filter = (
1200 Q(name__icontains=value) |
1201 Q(description__icontains=value)
1202 )
1203 return queryset.filter(qs_filter)
1206@register_filterset
1207class VLANTranslationRuleFilterSet(NetBoxModelFilterSet):
1208 policy_id = django_filters.ModelMultipleChoiceFilter(
1209 queryset=VLANTranslationPolicy.objects.all(),
1210 distinct=False,
1211 label=_('VLAN Translation Policy (ID)'),
1212 )
1213 policy = django_filters.ModelMultipleChoiceFilter(
1214 field_name='policy__name',
1215 queryset=VLANTranslationPolicy.objects.all(),
1216 distinct=False,
1217 to_field_name='name',
1218 label=_('VLAN Translation Policy (name)'),
1219 )
1221 class Meta:
1222 model = VLANTranslationRule
1223 fields = ('id', 'policy_id', 'policy', 'local_vid', 'remote_vid', 'description')
1225 def search(self, queryset, name, value):
1226 if not value.strip():
1227 return queryset
1228 qs_filter = (
1229 Q(policy__name__icontains=value)
1230 )
1231 try:
1232 int_value = int(value.strip())
1233 qs_filter |= Q(local_vid=int_value)
1234 qs_filter |= Q(remote_vid=int_value)
1235 except ValueError:
1236 pass
1237 return queryset.filter(qs_filter)
1240# Service/ServiceTemplate port filter name -> the port lookup it applies, in the order the conditions
1241# are built. See ServicePortMappingFilterMixin and ipam.utils.PORT_MAPPING_LOOKUPS.
1242SERVICE_PORT_FILTERS = {
1243 'port': 'exact',
1244 'port__gt': 'gt',
1245 'port__gte': 'gte',
1246 'port__lt': 'lt',
1247 'port__lte': 'lte',
1248}
1251class ServicePortMappingFilterMixin(django_filters.FilterSet):
1252 """
1253 Shared ``port_mappings``, ``protocol`` and ``port`` filtering for Service and ServiceTemplate, all
1254 operating on the ``port_mappings`` array. ``protocol`` and every active ``port`` lookup are
1255 correlated: they must all be satisfied by one single mapping, so ``?protocol=tcp&port__gt=1000`` does
1256 not match a service whose only tcp mapping is tcp/80, and ``?port__gte=1000&port__lte=2000`` does not
1257 match a service exposing only ports 500 and 5000. See ``ipam.utils.port_mapping_q``.
1258 """
1259 # Whole-mapping lookup, e.g. ?port_mappings=tcp/80. Each value already names one complete
1260 # protocol/port pair, so this needs none of the protocol/port correlation machinery below and is
1261 # simply ANDed with the other filters.
1262 port_mappings = MultiValueCharFilter(
1263 method='filter_port_mappings',
1264 label=_('Port mapping (protocol/port)'),
1265 )
1266 port_mappings__n = MultiValueCharFilter(
1267 method='filter_port_mappings_negated',
1268 label=_('Port mapping (protocol/port)'),
1269 )
1270 protocol = django_filters.MultipleChoiceFilter(
1271 choices=ServiceProtocolChoices,
1272 method='filter_protocol_noop',
1273 )
1274 # Negation lookup retained from when `protocol` was a model field: method-based filters don't get
1275 # the char-based lookups (protocol__n, __ic, ...) auto-generated, and silently dropping protocol__n
1276 # would widen existing saved filters/scripts rather than error. The __ic/__nic/__empty variants were
1277 # never meaningful on a small fixed choice set and are intentionally left gone.
1278 protocol__n = django_filters.MultipleChoiceFilter(
1279 choices=ServiceProtocolChoices,
1280 method='filter_protocol_negated',
1281 )
1282 # `port` and its range lookups. These are declared explicitly because a method-based filter gets no
1283 # auto-generated lookups (BaseFilterSet.get_additional_lookups() skips filters with a method), and
1284 # they must be correlated with `protocol` rather than applied independently. `port__empty` is
1285 # intentionally absent: port_mappings is never empty on a validated object, so it was never
1286 # meaningful. See ipam.utils.PORT_MAPPING_LOOKUPS for the lookup -> SQL operator mapping.
1287 #
1288 # `protocol` above and every `port*` lookup below (except the negations, which stand alone) are
1289 # deliberately no-ops: because they must be correlated with one another they cannot be applied as each
1290 # filter runs. filter_queryset() applies them together, once, after super() has applied the rest.
1291 port = MultiValueNumberFilter(
1292 method='filter_noop',
1293 )
1294 port__n = MultiValueNumberFilter(
1295 method='filter_port_negated',
1296 )
1297 port__gt = MultiValueNumberFilter(
1298 method='filter_noop',
1299 )
1300 port__gte = MultiValueNumberFilter(
1301 method='filter_noop',
1302 )
1303 port__lt = MultiValueNumberFilter(
1304 method='filter_noop',
1305 )
1306 port__lte = MultiValueNumberFilter(
1307 method='filter_noop',
1308 )
1310 def filter_queryset(self, queryset):
1311 """
1312 Apply `protocol` and every active `port*` lookup as a single correlated predicate.
1314 These can't be applied per-filter the way django-filter normally works: they must all be satisfied
1315 by one single mapping, and a query combining N of them would otherwise emit N independent (and
1316 redundant) copies of the same scan. So the individual filters are no-ops and the combined
1317 predicate is built here, from the cleaned data, exactly once per call.
1318 """
1319 queryset = super().filter_queryset(queryset)
1321 cleaned_data = self.form.cleaned_data
1322 protocols = cleaned_data.get('protocol') or []
1323 port_tests = [
1324 (lookup, values)
1325 for lookup, values in (
1326 (lookup, cleaned_data.get(name) or [])
1327 for name, lookup in SERVICE_PORT_FILTERS.items()
1328 )
1329 if values
1330 ]
1331 if not protocols and not port_tests: 1331 ↛ 1334line 1331 didn't jump to line 1334 because the condition on line 1331 was always true
1332 return queryset
1334 return queryset.filter(port_mapping_q(protocols, port_tests))
1336 def filter_noop(self, queryset, name, value):
1337 # See filter_queryset(), which applies `protocol` and the port lookups as one correlated predicate.
1338 return queryset
1340 def filter_protocol_noop(self, queryset, name, value: list[str]):
1341 # A no-op like filter_noop(), from which this differs only in its type hint. `protocol` is not a
1342 # model field, so the schema generator cannot infer the parameter's type from the model, and a
1343 # plain MultipleChoiceFilter carries none of the @extend_schema_field annotations which the
1344 # MultiValue* filters (used by the port lookups) do. The hint supplies it.
1345 return self.filter_noop(queryset, name, value)
1347 def filter_port_mappings(self, queryset, name, value: list[str]):
1348 # Array overlap (&&) is served by the GIN index on port_mappings and gives the multi-value OR
1349 # semantics used throughout NetBox: ?port_mappings=tcp/80&port_mappings=udp/53 matches either.
1350 if not value:
1351 return queryset
1352 return queryset.filter(port_mappings__overlap=[normalize_port_mapping(v) for v in value])
1354 def filter_port_mappings_negated(self, queryset, name, value: list[str]):
1355 if not value:
1356 return queryset
1357 return queryset.exclude(port_mappings__overlap=[normalize_port_mapping(v) for v in value])
1359 def filter_protocol_negated(self, queryset, name, value: list[str]):
1360 # Exclude services exposing any of the given protocols (negation of the protocol-only lookup).
1361 if not value:
1362 return queryset
1363 return queryset.exclude(port_mapping_q(value))
1365 def filter_port_negated(self, queryset, name, value: list[int]):
1366 # Exclude services exposing any of the given ports. Correlated with `protocol` when supplied, so
1367 # ?protocol=tcp&port__n=80 excludes only services exposing tcp/80 (not those exposing udp/80).
1368 if not value: 1368 ↛ 1369line 1368 didn't jump to line 1369 because the condition on line 1368 was never true
1369 return queryset
1370 protocols = self.form.cleaned_data.get('protocol') or []
1371 return queryset.exclude(port_mapping_q(protocols, [('exact', value)]))
1374@register_filterset
1375class ServiceTemplateFilterSet(ServicePortMappingFilterMixin, PrimaryModelFilterSet):
1377 class Meta:
1378 model = ServiceTemplate
1379 fields = ('id', 'name', 'description')
1381 def search(self, queryset, name, value):
1382 if not value.strip():
1383 return queryset
1384 qs_filter = (
1385 Q(name__icontains=value) |
1386 Q(description__icontains=value)
1387 )
1388 return queryset.filter(qs_filter)
1391@register_filterset
1392class ServiceFilterSet(ServicePortMappingFilterMixin, ContactModelFilterSet, PrimaryModelFilterSet):
1393 parent_object_type = MultiValueContentTypeFilter()
1394 device = MultiValueCharFilter(
1395 method='filter_device',
1396 field_name='name',
1397 label=_('Device (name)'),
1398 )
1399 device_id = MultiValueNumberFilter(
1400 method='filter_device',
1401 field_name='pk',
1402 label=_('Device (ID)'),
1403 )
1404 virtual_machine = MultiValueCharFilter(
1405 method='filter_virtual_machine',
1406 field_name='name',
1407 label=_('Virtual machine (name)'),
1408 )
1409 virtual_machine_id = MultiValueNumberFilter(
1410 method='filter_virtual_machine',
1411 field_name='pk',
1412 label=_('Virtual machine (ID)'),
1413 )
1414 fhrpgroup = MultiValueCharFilter(
1415 method='filter_fhrp_group',
1416 field_name='name',
1417 label=_('FHRP Group (name)'),
1418 )
1419 fhrpgroup_id = MultiValueNumberFilter(
1420 method='filter_fhrp_group',
1421 field_name='pk',
1422 label=_('FHRP Group (ID)'),
1423 )
1424 ip_address_id = django_filters.ModelMultipleChoiceFilter(
1425 field_name='ipaddresses',
1426 queryset=IPAddress.objects.all(),
1427 label=_('IP address (ID)'),
1428 )
1429 ip_address = django_filters.ModelMultipleChoiceFilter(
1430 field_name='ipaddresses__address',
1431 queryset=IPAddress.objects.all(),
1432 to_field_name='address',
1433 label=_('IP address'),
1434 )
1436 class Meta:
1437 model = Service
1438 fields = ('id', 'name', 'description', 'parent_object_type', 'parent_object_id')
1440 def search(self, queryset, name, value):
1441 if not value.strip():
1442 return queryset
1443 qs_filter = Q(name__icontains=value) | Q(description__icontains=value)
1444 return queryset.filter(qs_filter)
1446 def filter_device(self, queryset, name, value):
1447 devices = Device.objects.filter(**{'{}__in'.format(name): value})
1448 if not devices.exists():
1449 return queryset.none()
1450 service_ids = []
1451 for device in devices:
1452 service_ids.extend(device.services.values_list('id', flat=True))
1453 return queryset.filter(id__in=service_ids)
1455 def filter_fhrp_group(self, queryset, name, value):
1456 groups = FHRPGroup.objects.filter(**{'{}__in'.format(name): value})
1457 if not groups.exists():
1458 return queryset.none()
1459 service_ids = []
1460 for group in groups:
1461 service_ids.extend(group.services.values_list('id', flat=True))
1462 return queryset.filter(id__in=service_ids)
1464 def filter_virtual_machine(self, queryset, name, value):
1465 virtual_machines = VirtualMachine.objects.filter(**{'{}__in'.format(name): value})
1466 if not virtual_machines.exists():
1467 return queryset.none()
1468 service_ids = []
1469 for vm in virtual_machines:
1470 service_ids.extend(vm.services.values_list('id', flat=True))
1471 return queryset.filter(id__in=service_ids)
1474class PrimaryIPFilterSet(django_filters.FilterSet):
1475 """
1476 An inheritable FilterSet for models which support primary IP assignment.
1477 """
1478 primary_ip4_id = django_filters.ModelMultipleChoiceFilter(
1479 field_name='primary_ip4',
1480 queryset=IPAddress.objects.all(),
1481 distinct=False,
1482 label=_('Primary IPv4 (ID)'),
1483 )
1484 primary_ip4 = django_filters.ModelMultipleChoiceFilter(
1485 field_name='primary_ip4__address',
1486 queryset=IPAddress.objects.all(),
1487 distinct=False,
1488 to_field_name='address',
1489 label=_('Primary IPv4 (address)'),
1490 )
1491 primary_ip6_id = django_filters.ModelMultipleChoiceFilter(
1492 field_name='primary_ip6',
1493 queryset=IPAddress.objects.all(),
1494 distinct=False,
1495 label=_('Primary IPv6 (ID)'),
1496 )
1497 primary_ip6 = django_filters.ModelMultipleChoiceFilter(
1498 field_name='primary_ip6__address',
1499 queryset=IPAddress.objects.all(),
1500 distinct=False,
1501 to_field_name='address',
1502 label=_('Primary IPv6 (address)'),
1503 )