mirror of
https://github.com/netbox-community/netbox.git
synced 2024-05-10 07:54:54 +00:00
12180 available objects api (#12935)
* Introduce AvailableObjectsView and refactor 'available objects' API views * Restore advisory PostgreSQL locks * Move get_next_available_prefix() * Apply OpenAPI decorators for get() and post()
This commit is contained in:
Jeremy Stretch
GitHub
@ -58,6 +58,7 @@ class AvailableASNSerializer(serializers.Serializer):
|
||||
Representation of an ASN which does not exist in the database.
|
||||
"""
|
||||
asn = serializers.IntegerField(read_only=True)
|
||||
description = serializers.CharField(required=False)
|
||||
|
||||
def to_representation(self, asn):
|
||||
rir = NestedRIRSerializer(self.context['range'].rir, context={
|
||||
@ -432,6 +433,7 @@ class AvailableIPSerializer(serializers.Serializer):
|
||||
family = serializers.IntegerField(read_only=True)
|
||||
address = serializers.CharField(read_only=True)
|
||||
vrf = NestedVRFSerializer(read_only=True)
|
||||
description = serializers.CharField(required=False)
|
||||
|
||||
def to_representation(self, instance):
|
||||
if self.context.get('vrf'):
|
||||
|
||||
@ -3,7 +3,9 @@ from django.db import transaction
|
||||
from django.shortcuts import get_object_or_404
|
||||
from django_pglocks import advisory_lock
|
||||
from drf_spectacular.utils import extend_schema
|
||||
from netaddr import IPSet
|
||||
from rest_framework import status
|
||||
from rest_framework.exceptions import ValidationError
|
||||
from rest_framework.response import Response
|
||||
from rest_framework.routers import APIRootView
|
||||
from rest_framework.views import APIView
|
||||
@ -12,10 +14,12 @@ from circuits.models import Provider
|
||||
from dcim.models import Site
|
||||
from ipam import filtersets
|
||||
from ipam.models import *
|
||||
from ipam.utils import get_next_available_prefix
|
||||
from netbox.api.viewsets import NetBoxModelViewSet
|
||||
from netbox.api.viewsets.mixins import ObjectValidationMixin
|
||||
from netbox.config import get_config
|
||||
from netbox.constants import ADVISORY_LOCK_KEYS
|
||||
from utilities.api import get_serializer_for_model
|
||||
from utilities.utils import count_related
|
||||
from . import serializers
|
||||
from ipam.models import L2VPN, L2VPNTermination
|
||||
@ -207,237 +211,233 @@ def get_results_limit(request):
|
||||
return limit
|
||||
|
||||
|
||||
class AvailableASNsView(ObjectValidationMixin, APIView):
|
||||
queryset = ASN.objects.all()
|
||||
class AvailableObjectsView(ObjectValidationMixin, APIView):
|
||||
"""
|
||||
Return a list of dicts representing child objects that have not yet been created for a parent object.
|
||||
"""
|
||||
read_serializer_class = None
|
||||
write_serializer_class = None
|
||||
advisory_lock_key = None
|
||||
|
||||
def get_parent(self, request, pk):
|
||||
"""
|
||||
Return the parent object.
|
||||
"""
|
||||
raise NotImplemented()
|
||||
|
||||
def get_available_objects(self, parent, limit=None):
|
||||
"""
|
||||
Return all available objects for the parent.
|
||||
"""
|
||||
raise NotImplemented()
|
||||
|
||||
def get_extra_context(self, parent):
|
||||
"""
|
||||
Return any extra context data for the serializer.
|
||||
"""
|
||||
return {}
|
||||
|
||||
def check_sufficient_available(self, requested_objects, available_objects):
|
||||
"""
|
||||
Check if there exist a sufficient number of available objects to satisfy the request.
|
||||
"""
|
||||
return len(requested_objects) <= len(available_objects)
|
||||
|
||||
def prep_object_data(self, requested_objects, available_objects, parent):
|
||||
"""
|
||||
Prepare data by setting any programmatically determined object attributes (e.g. next available VLAN ID)
|
||||
on the request data.
|
||||
"""
|
||||
return requested_objects
|
||||
|
||||
@extend_schema(methods=["get"], responses={200: serializers.AvailableASNSerializer(many=True)})
|
||||
def get(self, request, pk):
|
||||
asnrange = get_object_or_404(ASNRange.objects.restrict(request.user), pk=pk)
|
||||
parent = self.get_parent(request, pk)
|
||||
limit = get_results_limit(request)
|
||||
available_objects = self.get_available_objects(parent, limit)
|
||||
|
||||
available_asns = asnrange.get_available_asns()[:limit]
|
||||
|
||||
serializer = serializers.AvailableASNSerializer(available_asns, many=True, context={
|
||||
serializer = self.read_serializer_class(available_objects, many=True, context={
|
||||
'request': request,
|
||||
'range': asnrange,
|
||||
**self.get_extra_context(parent),
|
||||
})
|
||||
|
||||
return Response(serializer.data)
|
||||
|
||||
@extend_schema(methods=["post"], responses={201: serializers.ASNSerializer(many=True)})
|
||||
@advisory_lock(ADVISORY_LOCK_KEYS['available-asns'])
|
||||
def post(self, request, pk):
|
||||
self.queryset = self.queryset.restrict(request.user, 'add')
|
||||
asnrange = get_object_or_404(ASNRange.objects.restrict(request.user), pk=pk)
|
||||
parent = self.get_parent(request, pk)
|
||||
|
||||
# Normalize to a list of objects
|
||||
requested_asns = request.data if isinstance(request.data, list) else [request.data]
|
||||
# Normalize request data to a list of objects
|
||||
requested_objects = request.data if isinstance(request.data, list) else [request.data]
|
||||
|
||||
# Determine if the requested number of IPs is available
|
||||
available_asns = asnrange.get_available_asns()
|
||||
if len(available_asns) < len(requested_asns):
|
||||
return Response(
|
||||
{
|
||||
"detail": f"An insufficient number of ASNs are available within {asnrange} "
|
||||
f"({len(requested_asns)} requested, {len(available_asns)} available)"
|
||||
},
|
||||
status=status.HTTP_409_CONFLICT
|
||||
)
|
||||
|
||||
# Assign ASNs from the list of available IPs and copy VRF assignment from the parent
|
||||
for i, requested_asn in enumerate(requested_asns):
|
||||
requested_asn.update({
|
||||
'rir': asnrange.rir.pk,
|
||||
'range': asnrange.pk,
|
||||
'asn': available_asns[i],
|
||||
})
|
||||
|
||||
# Initialize the serializer with a list or a single object depending on what was requested
|
||||
context = {'request': request}
|
||||
if isinstance(request.data, list):
|
||||
serializer = serializers.ASNSerializer(data=requested_asns, many=True, context=context)
|
||||
else:
|
||||
serializer = serializers.ASNSerializer(data=requested_asns[0], context=context)
|
||||
|
||||
# Create the new IP address(es)
|
||||
if serializer.is_valid():
|
||||
try:
|
||||
with transaction.atomic():
|
||||
created = serializer.save()
|
||||
self._validate_objects(created)
|
||||
except ObjectDoesNotExist:
|
||||
raise PermissionDenied()
|
||||
return Response(serializer.data, status=status.HTTP_201_CREATED)
|
||||
|
||||
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
def get_serializer_class(self):
|
||||
if self.request.method == "GET":
|
||||
return serializers.AvailableASNSerializer
|
||||
|
||||
return serializers.ASNSerializer
|
||||
|
||||
|
||||
class AvailablePrefixesView(ObjectValidationMixin, APIView):
|
||||
queryset = Prefix.objects.all()
|
||||
|
||||
@extend_schema(methods=["get"], responses={200: serializers.AvailablePrefixSerializer(many=True)})
|
||||
def get(self, request, pk):
|
||||
prefix = get_object_or_404(Prefix.objects.restrict(request.user), pk=pk)
|
||||
available_prefixes = prefix.get_available_prefixes()
|
||||
|
||||
serializer = serializers.AvailablePrefixSerializer(available_prefixes.iter_cidrs(), many=True, context={
|
||||
# Serialize and validate the request data
|
||||
serializer = self.write_serializer_class(data=requested_objects, many=True, context={
|
||||
'request': request,
|
||||
'vrf': prefix.vrf,
|
||||
**self.get_extra_context(parent),
|
||||
})
|
||||
|
||||
return Response(serializer.data)
|
||||
|
||||
@extend_schema(methods=["post"], responses={201: serializers.PrefixSerializer(many=True)})
|
||||
@advisory_lock(ADVISORY_LOCK_KEYS['available-prefixes'])
|
||||
def post(self, request, pk):
|
||||
self.queryset = self.queryset.restrict(request.user, 'add')
|
||||
prefix = get_object_or_404(Prefix.objects.restrict(request.user), pk=pk)
|
||||
available_prefixes = prefix.get_available_prefixes()
|
||||
|
||||
# Validate Requested Prefixes' length
|
||||
serializer = serializers.PrefixLengthSerializer(
|
||||
data=request.data if isinstance(request.data, list) else [request.data],
|
||||
many=True,
|
||||
context={
|
||||
'request': request,
|
||||
'prefix': prefix,
|
||||
}
|
||||
)
|
||||
if not serializer.is_valid():
|
||||
return Response(
|
||||
serializer.errors,
|
||||
status=status.HTTP_400_BAD_REQUEST
|
||||
)
|
||||
|
||||
requested_prefixes = serializer.validated_data
|
||||
# Allocate prefixes to the requested objects based on availability within the parent
|
||||
for i, requested_prefix in enumerate(requested_prefixes):
|
||||
with advisory_lock(ADVISORY_LOCK_KEYS[self.advisory_lock_key]):
|
||||
available_objects = self.get_available_objects(parent)
|
||||
|
||||
# Find the first available prefix equal to or larger than the requested size
|
||||
for available_prefix in available_prefixes.iter_cidrs():
|
||||
if requested_prefix['prefix_length'] >= available_prefix.prefixlen:
|
||||
allocated_prefix = '{}/{}'.format(available_prefix.network, requested_prefix['prefix_length'])
|
||||
requested_prefix['prefix'] = allocated_prefix
|
||||
requested_prefix['vrf'] = prefix.vrf.pk if prefix.vrf else None
|
||||
break
|
||||
else:
|
||||
# Determine if the requested number of objects is available
|
||||
if not self.check_sufficient_available(serializer.validated_data, available_objects):
|
||||
return Response(
|
||||
{
|
||||
"detail": "Insufficient space is available to accommodate the requested prefix size(s)"
|
||||
},
|
||||
{"detail": f"Insufficient resources are available to satisfy the request"},
|
||||
status=status.HTTP_409_CONFLICT
|
||||
)
|
||||
|
||||
# Remove the allocated prefix from the list of available prefixes
|
||||
available_prefixes.remove(allocated_prefix)
|
||||
# Prepare object data for deserialization
|
||||
requested_objects = self.prep_object_data(serializer.validated_data, available_objects, parent)
|
||||
|
||||
# Initialize the serializer with a list or a single object depending on what was requested
|
||||
context = {'request': request}
|
||||
if isinstance(request.data, list):
|
||||
serializer = serializers.PrefixSerializer(data=requested_prefixes, many=True, context=context)
|
||||
else:
|
||||
serializer = serializers.PrefixSerializer(data=requested_prefixes[0], context=context)
|
||||
# Initialize the serializer with a list or a single object depending on what was requested
|
||||
serializer_class = get_serializer_for_model(self.queryset.model)
|
||||
context = {'request': request}
|
||||
if isinstance(request.data, list):
|
||||
serializer = serializer_class(data=requested_objects, many=True, context=context)
|
||||
else:
|
||||
serializer = serializer_class(data=requested_objects[0], context=context)
|
||||
|
||||
# Create the new Prefix(es)
|
||||
if serializer.is_valid():
|
||||
if not serializer.is_valid():
|
||||
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
# Create the new IP address(es)
|
||||
try:
|
||||
with transaction.atomic():
|
||||
created = serializer.save()
|
||||
self._validate_objects(created)
|
||||
except ObjectDoesNotExist:
|
||||
raise PermissionDenied()
|
||||
return Response(serializer.data, status=status.HTTP_201_CREATED)
|
||||
|
||||
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
def get_serializer_class(self):
|
||||
if self.request.method == "GET":
|
||||
return serializers.AvailablePrefixSerializer
|
||||
|
||||
return serializers.PrefixLengthSerializer
|
||||
return Response(serializer.data, status=status.HTTP_201_CREATED)
|
||||
|
||||
|
||||
class AvailableIPAddressesView(ObjectValidationMixin, APIView):
|
||||
queryset = IPAddress.objects.all()
|
||||
class AvailableASNsView(AvailableObjectsView):
|
||||
queryset = ASN.objects.all()
|
||||
read_serializer_class = serializers.AvailableASNSerializer
|
||||
write_serializer_class = serializers.AvailableASNSerializer
|
||||
advisory_lock_key = 'available-asns'
|
||||
|
||||
def get_parent(self, request, pk):
|
||||
raise NotImplemented()
|
||||
return get_object_or_404(ASNRange.objects.restrict(request.user), pk=pk)
|
||||
|
||||
@extend_schema(methods=["get"], responses={200: serializers.AvailableIPSerializer(many=True)})
|
||||
def get_available_objects(self, parent, limit=None):
|
||||
return parent.get_available_asns()[:limit]
|
||||
|
||||
def get_extra_context(self, parent):
|
||||
return {
|
||||
'range': parent,
|
||||
}
|
||||
|
||||
def prep_object_data(self, requested_objects, available_objects, parent):
|
||||
for i, request_data in enumerate(requested_objects):
|
||||
request_data.update({
|
||||
'rir': parent.rir.pk,
|
||||
'range': parent.pk,
|
||||
'asn': available_objects[i],
|
||||
})
|
||||
|
||||
return requested_objects
|
||||
|
||||
@extend_schema(methods=["get"], responses={200: serializers.AvailableASNSerializer(many=True)})
|
||||
def get(self, request, pk):
|
||||
parent = self.get_parent(request, pk)
|
||||
limit = get_results_limit(request)
|
||||
return super().get(request, pk)
|
||||
|
||||
@extend_schema(methods=["post"], responses={201: serializers.AvailableASNSerializer(many=True)})
|
||||
def post(self, request, pk):
|
||||
return super().post(request, pk)
|
||||
|
||||
|
||||
class AvailablePrefixesView(AvailableObjectsView):
|
||||
queryset = Prefix.objects.all()
|
||||
read_serializer_class = serializers.AvailablePrefixSerializer
|
||||
write_serializer_class = serializers.PrefixLengthSerializer
|
||||
advisory_lock_key = 'available-prefixes'
|
||||
|
||||
def get_parent(self, request, pk):
|
||||
return get_object_or_404(Prefix.objects.restrict(request.user), pk=pk)
|
||||
|
||||
def get_available_objects(self, parent, limit=None):
|
||||
return parent.get_available_prefixes().iter_cidrs()
|
||||
|
||||
def check_sufficient_available(self, requested_objects, available_objects):
|
||||
available_prefixes = IPSet(available_objects)
|
||||
for requested_object in requested_objects:
|
||||
if not get_next_available_prefix(available_prefixes, requested_object['prefix_length']):
|
||||
return False
|
||||
return True
|
||||
|
||||
def get_extra_context(self, parent):
|
||||
return {
|
||||
'prefix': parent,
|
||||
'vrf': parent.vrf,
|
||||
}
|
||||
|
||||
def prep_object_data(self, requested_objects, available_objects, parent):
|
||||
available_prefixes = IPSet(available_objects)
|
||||
for i, request_data in enumerate(requested_objects):
|
||||
|
||||
# Find the first available prefix equal to or larger than the requested size
|
||||
if allocated_prefix := get_next_available_prefix(available_prefixes, request_data['prefix_length']):
|
||||
request_data.update({
|
||||
'prefix': allocated_prefix,
|
||||
'vrf': parent.vrf.pk if parent.vrf else None,
|
||||
})
|
||||
else:
|
||||
raise ValidationError("Insufficient space is available to accommodate the requested prefix size(s)")
|
||||
|
||||
return requested_objects
|
||||
|
||||
@extend_schema(methods=["get"], responses={200: serializers.AvailablePrefixSerializer(many=True)})
|
||||
def get(self, request, pk):
|
||||
return super().get(request, pk)
|
||||
|
||||
@extend_schema(methods=["post"], responses={201: serializers.PrefixSerializer(many=True)})
|
||||
def post(self, request, pk):
|
||||
return super().post(request, pk)
|
||||
|
||||
|
||||
class AvailableIPAddressesView(AvailableObjectsView):
|
||||
queryset = IPAddress.objects.all()
|
||||
read_serializer_class = serializers.AvailableIPSerializer
|
||||
write_serializer_class = serializers.AvailableIPSerializer
|
||||
advisory_lock_key = 'available-ips'
|
||||
|
||||
def get_available_objects(self, parent, limit=None):
|
||||
# Calculate available IPs within the parent
|
||||
ip_list = []
|
||||
for index, ip in enumerate(parent.get_available_ips(), start=1):
|
||||
ip_list.append(ip)
|
||||
if index == limit:
|
||||
break
|
||||
serializer = serializers.AvailableIPSerializer(ip_list, many=True, context={
|
||||
'request': request,
|
||||
return ip_list
|
||||
|
||||
def get_extra_context(self, parent):
|
||||
return {
|
||||
'parent': parent,
|
||||
'vrf': parent.vrf,
|
||||
})
|
||||
}
|
||||
|
||||
return Response(serializer.data)
|
||||
def prep_object_data(self, requested_objects, available_objects, parent):
|
||||
available_ips = iter(available_objects)
|
||||
for i, request_data in enumerate(requested_objects):
|
||||
request_data.update({
|
||||
'address': f'{next(available_ips)}/{parent.mask_length}',
|
||||
'vrf': parent.vrf.pk if parent.vrf else None,
|
||||
})
|
||||
|
||||
return requested_objects
|
||||
|
||||
@extend_schema(methods=["get"], responses={200: serializers.AvailableIPSerializer(many=True)})
|
||||
def get(self, request, pk):
|
||||
return super().get(request, pk)
|
||||
|
||||
@extend_schema(methods=["post"], responses={201: serializers.IPAddressSerializer(many=True)})
|
||||
@advisory_lock(ADVISORY_LOCK_KEYS['available-ips'])
|
||||
def post(self, request, pk):
|
||||
self.queryset = self.queryset.restrict(request.user, 'add')
|
||||
parent = self.get_parent(request, pk)
|
||||
|
||||
# Normalize to a list of objects
|
||||
requested_ips = request.data if isinstance(request.data, list) else [request.data]
|
||||
|
||||
# Determine if the requested number of IPs is available
|
||||
available_ips = parent.get_available_ips()
|
||||
if available_ips.size < len(requested_ips):
|
||||
return Response(
|
||||
{
|
||||
"detail": f"An insufficient number of IP addresses are available within {parent} "
|
||||
f"({len(requested_ips)} requested, {len(available_ips)} available)"
|
||||
},
|
||||
status=status.HTTP_409_CONFLICT
|
||||
)
|
||||
|
||||
# Assign addresses from the list of available IPs and copy VRF assignment from the parent
|
||||
available_ips = iter(available_ips)
|
||||
for requested_ip in requested_ips:
|
||||
requested_ip['address'] = f'{next(available_ips)}/{parent.mask_length}'
|
||||
requested_ip['vrf'] = parent.vrf.pk if parent.vrf else None
|
||||
|
||||
# Initialize the serializer with a list or a single object depending on what was requested
|
||||
context = {'request': request}
|
||||
if isinstance(request.data, list):
|
||||
serializer = serializers.IPAddressSerializer(data=requested_ips, many=True, context=context)
|
||||
else:
|
||||
serializer = serializers.IPAddressSerializer(data=requested_ips[0], context=context)
|
||||
|
||||
# Create the new IP address(es)
|
||||
if serializer.is_valid():
|
||||
try:
|
||||
with transaction.atomic():
|
||||
created = serializer.save()
|
||||
self._validate_objects(created)
|
||||
except ObjectDoesNotExist:
|
||||
raise PermissionDenied()
|
||||
return Response(serializer.data, status=status.HTTP_201_CREATED)
|
||||
|
||||
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
def get_serializer_class(self):
|
||||
if self.request.method == "GET":
|
||||
return serializers.AvailableIPSerializer
|
||||
|
||||
return serializers.IPAddressSerializer
|
||||
return super().post(request, pk)
|
||||
|
||||
|
||||
class PrefixAvailableIPAddressesView(AvailableIPAddressesView):
|
||||
@ -452,77 +452,36 @@ class IPRangeAvailableIPAddressesView(AvailableIPAddressesView):
|
||||
return get_object_or_404(IPRange.objects.restrict(request.user), pk=pk)
|
||||
|
||||
|
||||
class AvailableVLANsView(ObjectValidationMixin, APIView):
|
||||
class AvailableVLANsView(AvailableObjectsView):
|
||||
queryset = VLAN.objects.all()
|
||||
read_serializer_class = serializers.AvailableVLANSerializer
|
||||
write_serializer_class = serializers.CreateAvailableVLANSerializer
|
||||
advisory_lock_key = 'available-vlans'
|
||||
|
||||
def get_parent(self, request, pk):
|
||||
return get_object_or_404(VLANGroup.objects.restrict(request.user), pk=pk)
|
||||
|
||||
def get_available_objects(self, parent, limit=None):
|
||||
return parent.get_available_vids()[:limit]
|
||||
|
||||
def get_extra_context(self, parent):
|
||||
return {
|
||||
'group': parent,
|
||||
}
|
||||
|
||||
def prep_object_data(self, requested_objects, available_objects, parent):
|
||||
for i, request_data in enumerate(requested_objects):
|
||||
request_data.update({
|
||||
'vid': available_objects.pop(0),
|
||||
'group': parent.pk,
|
||||
})
|
||||
|
||||
return requested_objects
|
||||
|
||||
@extend_schema(methods=["get"], responses={200: serializers.AvailableVLANSerializer(many=True)})
|
||||
def get(self, request, pk):
|
||||
vlangroup = get_object_or_404(VLANGroup.objects.restrict(request.user), pk=pk)
|
||||
limit = get_results_limit(request)
|
||||
|
||||
available_vlans = vlangroup.get_available_vids()[:limit]
|
||||
serializer = serializers.AvailableVLANSerializer(available_vlans, many=True, context={
|
||||
'request': request,
|
||||
'group': vlangroup,
|
||||
})
|
||||
|
||||
return Response(serializer.data)
|
||||
return super().get(request, pk)
|
||||
|
||||
@extend_schema(methods=["post"], responses={201: serializers.VLANSerializer(many=True)})
|
||||
@advisory_lock(ADVISORY_LOCK_KEYS['available-vlans'])
|
||||
def post(self, request, pk):
|
||||
self.queryset = self.queryset.restrict(request.user, 'add')
|
||||
vlangroup = get_object_or_404(VLANGroup.objects.restrict(request.user), pk=pk)
|
||||
available_vlans = vlangroup.get_available_vids()
|
||||
many = isinstance(request.data, list)
|
||||
|
||||
# Validate requested VLANs
|
||||
serializer = serializers.CreateAvailableVLANSerializer(
|
||||
data=request.data if many else [request.data],
|
||||
many=True,
|
||||
context={
|
||||
'request': request,
|
||||
'group': vlangroup,
|
||||
}
|
||||
)
|
||||
if not serializer.is_valid():
|
||||
return Response(
|
||||
serializer.errors,
|
||||
status=status.HTTP_400_BAD_REQUEST
|
||||
)
|
||||
|
||||
requested_vlans = serializer.validated_data
|
||||
|
||||
for i, requested_vlan in enumerate(requested_vlans):
|
||||
try:
|
||||
requested_vlan['vid'] = available_vlans.pop(0)
|
||||
requested_vlan['group'] = vlangroup.pk
|
||||
except IndexError:
|
||||
return Response({
|
||||
"detail": "The requested number of VLANs is not available"
|
||||
}, status=status.HTTP_409_CONFLICT)
|
||||
|
||||
# Initialize the serializer with a list or a single object depending on what was requested
|
||||
context = {'request': request}
|
||||
if many:
|
||||
serializer = serializers.VLANSerializer(data=requested_vlans, many=True, context=context)
|
||||
else:
|
||||
serializer = serializers.VLANSerializer(data=requested_vlans[0], context=context)
|
||||
|
||||
# Create the new VLAN(s)
|
||||
if serializer.is_valid():
|
||||
try:
|
||||
with transaction.atomic():
|
||||
created = serializer.save()
|
||||
self._validate_objects(created)
|
||||
except ObjectDoesNotExist:
|
||||
raise PermissionDenied()
|
||||
return Response(serializer.data, status=status.HTTP_201_CREATED)
|
||||
|
||||
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
def get_serializer_class(self):
|
||||
if self.request.method == "GET":
|
||||
return serializers.AvailableVLANSerializer
|
||||
|
||||
return serializers.VLANSerializer
|
||||
return super().post(request, pk)
|
||||
|
||||
@ -1,7 +1,15 @@
|
||||
import netaddr
|
||||
|
||||
from .constants import *
|
||||
from .models import ASN, Prefix, VLAN
|
||||
from .models import Prefix, VLAN
|
||||
|
||||
__all__ = (
|
||||
'add_available_ipaddresses',
|
||||
'add_available_vlans',
|
||||
'add_requested_prefixes',
|
||||
'get_next_available_prefix',
|
||||
'rebuild_prefixes',
|
||||
)
|
||||
|
||||
|
||||
def add_requested_prefixes(parent, prefix_list, show_available=True, show_assigned=True):
|
||||
@ -184,3 +192,15 @@ def rebuild_prefixes(vrf):
|
||||
|
||||
# Final flush of any remaining Prefixes
|
||||
Prefix.objects.bulk_update(update_queue, ['_depth', '_children'])
|
||||
|
||||
|
||||
def get_next_available_prefix(ipset, prefix_size):
|
||||
"""
|
||||
Given a prefix length, allocate the next available prefix from an IPSet.
|
||||
"""
|
||||
for available_prefix in ipset.iter_cidrs():
|
||||
if prefix_size >= available_prefix.prefixlen:
|
||||
allocated_prefix = f"{available_prefix.network}/{prefix_size}"
|
||||
ipset.remove(allocated_prefix)
|
||||
return allocated_prefix
|
||||
return None
|
||||
|
||||
Reference in New Issue
Block a user