mirror/netbox-community-netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2024-05-10 07:54:54 +00:00
Code Activity

Reorganize REST API components under netbox app

Browse Source
This commit is contained in:
Jeremy Stretch
2020-10-13 15:54:23 -04:00
parent 97ac6851d0
commit c0c5f52ed9
47 changed files with 753 additions and 750 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
netbox/circuits/api/nested_serializers.py
View File

@ -1,7 +1,7 @@
from rest_framework import serializers
from circuits.models import Circuit, CircuitTermination, CircuitType, Provider
from utilities.api import WritableNestedSerializer
from netbox.api import WritableNestedSerializer
__all__ = [
'NestedCircuitSerializer',

2
netbox/circuits/api/serializers.py
View File

@ -6,8 +6,8 @@ from dcim.api.nested_serializers import NestedCableSerializer, NestedInterfaceSe
from dcim.api.serializers import CableTerminationSerializer, ConnectedEndpointSerializer
from extras.api.customfields import CustomFieldModelSerializer
from extras.api.serializers import TaggedObjectSerializer
from netbox.api import ChoiceField, ValidatedModelSerializer, WritableNestedSerializer
from tenancy.api.nested_serializers import NestedTenantSerializer
from utilities.api import ChoiceField, ValidatedModelSerializer, WritableNestedSerializer
from .nested_serializers import *

2
netbox/circuits/api/urls.py
View File

@ -1,4 +1,4 @@
from utilities.api import OrderedDefaultRouter
from netbox.api import OrderedDefaultRouter
from . import views

2
netbox/circuits/api/views.py
View File

@ -5,7 +5,7 @@ from circuits import filters
from circuits.models import Provider, CircuitTermination, CircuitType, Circuit
from dcim.api.views import PathEndpointMixin
from extras.api.views import CustomFieldModelViewSet
from utilities.api import ModelViewSet
from netbox.api.views import ModelViewSet
from . import serializers

2
netbox/dcim/api/nested_serializers.py
View File

@ -1,7 +1,7 @@
from rest_framework import serializers
from dcim import models
from utilities.api import WritableNestedSerializer
from netbox.api import WritableNestedSerializer
__all__ = [
'NestedCableSerializer',

9
netbox/dcim/api/serializers.py
View File

@ -18,12 +18,13 @@ from extras.api.customfields import CustomFieldModelSerializer
from extras.api.serializers import TaggedObjectSerializer
from ipam.api.nested_serializers import NestedIPAddressSerializer, NestedVLANSerializer
from ipam.models import VLAN
from netbox.api import (
ChoiceField, ContentTypeField, SerializedPKRelatedField, TimeZoneField, ValidatedModelSerializer,
WritableNestedSerializer,
)
from tenancy.api.nested_serializers import NestedTenantSerializer
from users.api.nested_serializers import NestedUserSerializer
from utilities.api import (
ChoiceField, ContentTypeField, SerializedPKRelatedField, TimeZoneField, ValidatedModelSerializer,
WritableNestedSerializer, get_serializer_for_model,
)
from utilities.api import get_serializer_for_model
from virtualization.api.nested_serializers import NestedClusterSerializer
from .nested_serializers import *

2
netbox/dcim/api/urls.py
View File

@ -1,4 +1,4 @@
from utilities.api import OrderedDefaultRouter
from netbox.api import OrderedDefaultRouter
from . import views

9
netbox/dcim/api/views.py
View File

@ -25,11 +25,12 @@ from dcim.models import (
)
from extras.api.views import CustomFieldModelViewSet
from ipam.models import Prefix, VLAN
from utilities.api import (
get_serializer_for_model, IsAuthenticatedOrLoginNotRequired, ModelViewSet, ServiceUnavailable,
)
from netbox.api.views import ModelViewSet
from netbox.api.authentication import IsAuthenticatedOrLoginNotRequired
from netbox.api.exceptions import ServiceUnavailable
from netbox.api.metadata import ContentTypeMetadata
from utilities.api import get_serializer_for_model
from utilities.utils import get_subquery
from utilities.metadata import ContentTypeMetadata
from virtualization.models import VirtualMachine
from . import serializers
from .exceptions import MissingFilterException

2
netbox/extras/api/customfields.py
View File

@ -6,7 +6,7 @@ from rest_framework.fields import CreateOnlyDefault, Field
from extras.choices import *
from extras.models import CustomField
from utilities.api import ValidatedModelSerializer
from netbox.api import ValidatedModelSerializer
#

2
netbox/extras/api/nested_serializers.py
View File

@ -1,8 +1,8 @@
from rest_framework import serializers
from extras import choices, models
from netbox.api import ChoiceField, WritableNestedSerializer
from users.api.nested_serializers import NestedUserSerializer
from utilities.api import ChoiceField, WritableNestedSerializer
__all__ = [
'NestedConfigContextSerializer',

7
netbox/extras/api/serializers.py
View File

@ -13,13 +13,12 @@ from extras.models import (
ConfigContext, ExportTemplate, ImageAttachment, ObjectChange, JobResult, Tag,
)
from extras.utils import FeatureQuery
from netbox.api import ChoiceField, ContentTypeField, SerializedPKRelatedField, ValidatedModelSerializer
from netbox.api.exceptions import SerializerNotFound
from tenancy.api.nested_serializers import NestedTenantSerializer, NestedTenantGroupSerializer
from tenancy.models import Tenant, TenantGroup
from users.api.nested_serializers import NestedUserSerializer
from utilities.api import (
ChoiceField, ContentTypeField, get_serializer_for_model, SerializerNotFound, SerializedPKRelatedField,
ValidatedModelSerializer,
)
from utilities.api import get_serializer_for_model
from virtualization.api.nested_serializers import NestedClusterGroupSerializer, NestedClusterSerializer
from virtualization.models import Cluster, ClusterGroup
from .nested_serializers import *

2
netbox/extras/api/urls.py
View File

@ -1,4 +1,4 @@
from utilities.api import OrderedDefaultRouter
from netbox.api import OrderedDefaultRouter
from . import views

5
netbox/extras/api/views.py
View File

@ -15,9 +15,10 @@ from extras.choices import JobResultStatusChoices
from extras.models import ConfigContext, ExportTemplate, ImageAttachment, ObjectChange, JobResult, Tag
from extras.reports import get_report, get_reports, run_report
from extras.scripts import get_script, get_scripts, run_script
from utilities.api import IsAuthenticatedOrLoginNotRequired, ModelViewSet
from netbox.api.views import ModelViewSet
from netbox.api.authentication import IsAuthenticatedOrLoginNotRequired
from netbox.api.metadata import ContentTypeMetadata
from utilities.exceptions import RQWorkerNotRunningException
from utilities.metadata import ContentTypeMetadata
from utilities.utils import copy_safe_request
from . import serializers

2
netbox/ipam/api/nested_serializers.py
View File

@ -1,7 +1,7 @@
from rest_framework import serializers
from ipam import models
from utilities.api import WritableNestedSerializer
from netbox.api import WritableNestedSerializer
__all__ = [
'NestedAggregateSerializer',

5
netbox/ipam/api/serializers.py
View File

@ -11,10 +11,9 @@ from extras.api.serializers import TaggedObjectSerializer
from ipam.choices import *
from ipam.constants import IPADDRESS_ASSIGNMENT_MODELS
from ipam.models import Aggregate, IPAddress, Prefix, RIR, Role, RouteTarget, Service, VLAN, VLANGroup, VRF
from netbox.api import ChoiceField, ContentTypeField, SerializedPKRelatedField, ValidatedModelSerializer
from tenancy.api.nested_serializers import NestedTenantSerializer
from utilities.api import (
ChoiceField, ContentTypeField, SerializedPKRelatedField, ValidatedModelSerializer, get_serializer_for_model,
)
from utilities.api import get_serializer_for_model
from virtualization.api.nested_serializers import NestedVirtualMachineSerializer
from .nested_serializers import *

2
netbox/ipam/api/urls.py
View File

@ -1,4 +1,4 @@
from utilities.api import OrderedDefaultRouter
from netbox.api import OrderedDefaultRouter
from . import views

2
netbox/ipam/api/views.py
View File

@ -11,7 +11,7 @@ from rest_framework.routers import APIRootView
from extras.api.views import CustomFieldModelViewSet
from ipam import filters
from ipam.models import Aggregate, IPAddress, Prefix, RIR, Role, RouteTarget, Service, VLAN, VLANGroup, VRF
from utilities.api import ModelViewSet
from netbox.api.views import ModelViewSet
from utilities.constants import ADVISORY_LOCK_KEYS
from utilities.utils import get_subquery
from . import serializers

207
netbox/netbox/api.py
View File

@ -1,207 +0,0 @@
from django.conf import settings
from django.db.models import QuerySet
from rest_framework import authentication, exceptions
from rest_framework.pagination import LimitOffsetPagination
from rest_framework.permissions import DjangoObjectPermissions, SAFE_METHODS
from rest_framework.renderers import BrowsableAPIRenderer
from rest_framework.schemas import coreapi
from rest_framework.utils import formatting
from users.models import Token
def is_custom_action(action):
return action not in {
# Default actions
'retrieve', 'list', 'create', 'update', 'partial_update', 'destroy',
# Bulk operations
'bulk_update', 'bulk_partial_update', 'bulk_destroy',
}
# Monkey-patch DRF to treat bulk_destroy() as a non-custom action (see #3436)
coreapi.is_custom_action = is_custom_action
#
# Renderers
#
class FormlessBrowsableAPIRenderer(BrowsableAPIRenderer):
"""
Override the built-in BrowsableAPIRenderer to disable HTML forms.
"""
def show_form_for_method(self, *args, **kwargs):
return False
def get_filter_form(self, data, view, request):
return None
#
# Authentication
#
class TokenAuthentication(authentication.TokenAuthentication):
"""
A custom authentication scheme which enforces Token expiration times.
"""
model = Token
def authenticate_credentials(self, key):
model = self.get_model()
try:
token = model.objects.prefetch_related('user').get(key=key)
except model.DoesNotExist:
raise exceptions.AuthenticationFailed("Invalid token")
# Enforce the Token's expiration time, if one has been set.
if token.is_expired:
raise exceptions.AuthenticationFailed("Token expired")
if not token.user.is_active:
raise exceptions.AuthenticationFailed("User inactive")
return token.user, token
class TokenPermissions(DjangoObjectPermissions):
"""
Custom permissions handler which extends the built-in DjangoModelPermissions to validate a Token's write ability
for unsafe requests (POST/PUT/PATCH/DELETE).
"""
# Override the stock perm_map to enforce view permissions
perms_map = {
'GET': ['%(app_label)s.view_%(model_name)s'],
'OPTIONS': [],
'HEAD': ['%(app_label)s.view_%(model_name)s'],
'POST': ['%(app_label)s.add_%(model_name)s'],
'PUT': ['%(app_label)s.change_%(model_name)s'],
'PATCH': ['%(app_label)s.change_%(model_name)s'],
'DELETE': ['%(app_label)s.delete_%(model_name)s'],
}
def __init__(self):
# LOGIN_REQUIRED determines whether read-only access is provided to anonymous users.
self.authenticated_users_only = settings.LOGIN_REQUIRED
super().__init__()
def _verify_write_permission(self, request):
# If token authentication is in use, verify that the token allows write operations (for unsafe methods).
if request.method in SAFE_METHODS or request.auth.write_enabled:
return True
def has_permission(self, request, view):
# Enforce Token write ability
if isinstance(request.auth, Token) and not self._verify_write_permission(request):
return False
return super().has_permission(request, view)
def has_object_permission(self, request, view, obj):
# Enforce Token write ability
if isinstance(request.auth, Token) and not self._verify_write_permission(request):
return False
return super().has_object_permission(request, view, obj)
#
# Pagination
#
class OptionalLimitOffsetPagination(LimitOffsetPagination):
"""
Override the stock paginator to allow setting limit=0 to disable pagination for a request. This returns all objects
matching a query, but retains the same format as a paginated request. The limit can only be disabled if
MAX_PAGE_SIZE has been set to 0 or None.
"""
def paginate_queryset(self, queryset, request, view=None):
if isinstance(queryset, QuerySet):
self.count = queryset.count()
else:
# We're dealing with an iterable, not a QuerySet
self.count = len(queryset)
self.limit = self.get_limit(request)
self.offset = self.get_offset(request)
self.request = request
if self.limit and self.count > self.limit and self.template is not None:
self.display_page_controls = True
if self.count == 0 or self.offset > self.count:
return list()
if self.limit:
return list(queryset[self.offset:self.offset + self.limit])
else:
return list(queryset[self.offset:])
def get_limit(self, request):
if self.limit_query_param:
try:
limit = int(request.query_params[self.limit_query_param])
if limit < 0:
raise ValueError()
# Enforce maximum page size, if defined
if settings.MAX_PAGE_SIZE:
if limit == 0:
return settings.MAX_PAGE_SIZE
else:
return min(limit, settings.MAX_PAGE_SIZE)
return limit
except (KeyError, ValueError):
pass
return self.default_limit
def get_next_link(self):
# Pagination has been disabled
if not self.limit:
return None
return super().get_next_link()
def get_previous_link(self):
# Pagination has been disabled
if not self.limit:
return None
return super().get_previous_link()
#
# Miscellaneous
#
def get_view_name(view, suffix=None):
"""
Derive the view name from its associated model, if it has one. Fall back to DRF's built-in `get_view_name`.
"""
if hasattr(view, 'queryset'):
# Determine the model name from the queryset.
name = view.queryset.model._meta.verbose_name
name = ' '.join([w[0].upper() + w[1:] for w in name.split()]) # Capitalize each word
else:
# Replicate DRF's built-in behavior.
name = view.__class__.__name__
name = formatting.remove_trailing_string(name, 'View')
name = formatting.remove_trailing_string(name, 'ViewSet')
name = formatting.camelcase_to_spaces(name)
if suffix:
name += ' ' + suffix
return name

30
netbox/netbox/api/__init__.py Normal file
View File

@ -0,0 +1,30 @@
from rest_framework.schemas import coreapi
from .fields import ChoiceField, ContentTypeField, SerializedPKRelatedField, TimeZoneField
from .routers import OrderedDefaultRouter
from .serializers import BulkOperationSerializer, ValidatedModelSerializer, WritableNestedSerializer
__all__ = (
'BulkOperationSerializer',
'ChoiceField',
'ContentTypeField',
'OrderedDefaultRouter',
'SerializedPKRelatedField',
'TimeZoneField',
'ValidatedModelSerializer',
'WritableNestedSerializer',
)
def is_custom_action(action):
return action not in {
# Default actions
'retrieve', 'list', 'create', 'update', 'partial_update', 'destroy',
# Bulk operations
'bulk_update', 'bulk_partial_update', 'bulk_destroy',
}
# Monkey-patch DRF to treat bulk_destroy() as a non-custom action (see #3436)
coreapi.is_custom_action = is_custom_action

84
netbox/netbox/api/authentication.py Normal file
View File

@ -0,0 +1,84 @@
from django.conf import settings
from rest_framework import authentication, exceptions
from rest_framework.permissions import BasePermission, DjangoObjectPermissions, SAFE_METHODS
from users.models import Token
class TokenAuthentication(authentication.TokenAuthentication):
"""
A custom authentication scheme which enforces Token expiration times.
"""
model = Token
def authenticate_credentials(self, key):
model = self.get_model()
try:
token = model.objects.prefetch_related('user').get(key=key)
except model.DoesNotExist:
raise exceptions.AuthenticationFailed("Invalid token")
# Enforce the Token's expiration time, if one has been set.
if token.is_expired:
raise exceptions.AuthenticationFailed("Token expired")
if not token.user.is_active:
raise exceptions.AuthenticationFailed("User inactive")
return token.user, token
class TokenPermissions(DjangoObjectPermissions):
"""
Custom permissions handler which extends the built-in DjangoModelPermissions to validate a Token's write ability
for unsafe requests (POST/PUT/PATCH/DELETE).
"""
# Override the stock perm_map to enforce view permissions
perms_map = {
'GET': ['%(app_label)s.view_%(model_name)s'],
'OPTIONS': [],
'HEAD': ['%(app_label)s.view_%(model_name)s'],
'POST': ['%(app_label)s.add_%(model_name)s'],
'PUT': ['%(app_label)s.change_%(model_name)s'],
'PATCH': ['%(app_label)s.change_%(model_name)s'],
'DELETE': ['%(app_label)s.delete_%(model_name)s'],
}
def __init__(self):
# LOGIN_REQUIRED determines whether read-only access is provided to anonymous users.
self.authenticated_users_only = settings.LOGIN_REQUIRED
super().__init__()
def _verify_write_permission(self, request):
# If token authentication is in use, verify that the token allows write operations (for unsafe methods).
if request.method in SAFE_METHODS or request.auth.write_enabled:
return True
def has_permission(self, request, view):
# Enforce Token write ability
if isinstance(request.auth, Token) and not self._verify_write_permission(request):
return False
return super().has_permission(request, view)
def has_object_permission(self, request, view, obj):
# Enforce Token write ability
if isinstance(request.auth, Token) and not self._verify_write_permission(request):
return False
return super().has_object_permission(request, view, obj)
class IsAuthenticatedOrLoginNotRequired(BasePermission):
"""
Returns True if the user is authenticated or LOGIN_REQUIRED is False.
"""
def has_permission(self, request, view):
if not settings.LOGIN_REQUIRED:
return True
return request.user.is_authenticated

10
netbox/netbox/api/exceptions.py Normal file
View File

@ -0,0 +1,10 @@
from rest_framework.exceptions import APIException
class ServiceUnavailable(APIException):
status_code = 503
default_detail = "Service temporarily unavailable, please try again later."
class SerializerNotFound(Exception):
pass

133
netbox/netbox/api/fields.py Normal file
View File

@ -0,0 +1,133 @@
from collections import OrderedDict
import pytz
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ObjectDoesNotExist
from rest_framework import serializers
from rest_framework.exceptions import ValidationError
from rest_framework.relations import PrimaryKeyRelatedField, RelatedField
class ChoiceField(serializers.Field):
"""
Represent a ChoiceField as {'value': <DB value>, 'label': <string>}. Accepts a single value on write.
:param choices: An iterable of choices in the form (value, key).
:param allow_blank: Allow blank values in addition to the listed choices.
"""
def __init__(self, choices, allow_blank=False, **kwargs):
self.choiceset = choices
self.allow_blank = allow_blank
self._choices = dict()
# Unpack grouped choices
for k, v in choices:
if type(v) in [list, tuple]:
for k2, v2 in v:
self._choices[k2] = v2
else:
self._choices[k] = v
super().__init__(**kwargs)
def validate_empty_values(self, data):
# Convert null to an empty string unless allow_null == True
if data is None:
if self.allow_null:
return True, None
else:
data = ''
return super().validate_empty_values(data)
def to_representation(self, obj):
if obj is '':
return None
return OrderedDict([
('value', obj),
('label', self._choices[obj])
])
def to_internal_value(self, data):
if data is '':
if self.allow_blank:
return data
raise ValidationError("This field may not be blank.")
# Provide an explicit error message if the request is trying to write a dict or list
if isinstance(data, (dict, list)):
raise ValidationError('Value must be passed directly (e.g. "foo": 123); do not use a dictionary or list.')
# Check for string representations of boolean/integer values
if hasattr(data, 'lower'):
if data.lower() == 'true':
data = True
elif data.lower() == 'false':
data = False
else:
try:
data = int(data)
except ValueError:
pass
try:
if data in self._choices:
return data
except TypeError: # Input is an unhashable type
pass
raise ValidationError(f"{data} is not a valid choice.")
@property
def choices(self):
return self._choices
class ContentTypeField(RelatedField):
"""
Represent a ContentType as '<app_label>.<model>'
"""
default_error_messages = {
"does_not_exist": "Invalid content type: {content_type}",
"invalid": "Invalid value. Specify a content type as '<app_label>.<model_name>'.",
}
def to_internal_value(self, data):
try:
app_label, model = data.split('.')
return ContentType.objects.get_by_natural_key(app_label=app_label, model=model)
except ObjectDoesNotExist:
self.fail('does_not_exist', content_type=data)
except (TypeError, ValueError):
self.fail('invalid')
def to_representation(self, obj):
return "{}.{}".format(obj.app_label, obj.model)
class TimeZoneField(serializers.Field):
"""
Represent a pytz time zone.
"""
def to_representation(self, obj):
return obj.zone if obj else None
def to_internal_value(self, data):
if not data:
return ""
if data not in pytz.common_timezones:
raise ValidationError('Unknown time zone "{}" (see pytz.common_timezones for all options)'.format(data))
return pytz.timezone(data)
class SerializedPKRelatedField(PrimaryKeyRelatedField):
"""
Extends PrimaryKeyRelatedField to return a serialized object on read. This is useful for representing related
objects in a ManyToManyField while still allowing a set of primary keys to be written.
"""
def __init__(self, serializer, **kwargs):
self.serializer = serializer
self.pk_field = kwargs.pop('pk_field', None)
super().__init__(**kwargs)
def to_representation(self, value):
return self.serializer(value, context={'request': self.context['request']}).data

5
netbox/utilities/metadata.py → netbox/netbox/api/metadata.py
View File

@ -1,6 +1,7 @@
from rest_framework.metadata import SimpleMetadata
from django.utils.encoding import force_str
from utilities.api import ContentTypeField
from rest_framework.metadata import SimpleMetadata
from netbox.api import ContentTypeField
class ContentTypeMetadata(SimpleMetadata):

69
netbox/netbox/api/pagination.py Normal file
View File

@ -0,0 +1,69 @@
from django.conf import settings
from django.db.models import QuerySet
from rest_framework.pagination import LimitOffsetPagination
class OptionalLimitOffsetPagination(LimitOffsetPagination):
"""
Override the stock paginator to allow setting limit=0 to disable pagination for a request. This returns all objects
matching a query, but retains the same format as a paginated request. The limit can only be disabled if
MAX_PAGE_SIZE has been set to 0 or None.
"""
def paginate_queryset(self, queryset, request, view=None):
if isinstance(queryset, QuerySet):
self.count = queryset.count()
else:
# We're dealing with an iterable, not a QuerySet
self.count = len(queryset)
self.limit = self.get_limit(request)
self.offset = self.get_offset(request)
self.request = request
if self.limit and self.count > self.limit and self.template is not None:
self.display_page_controls = True
if self.count == 0 or self.offset > self.count:
return list()
if self.limit:
return list(queryset[self.offset:self.offset + self.limit])
else:
return list(queryset[self.offset:])
def get_limit(self, request):
if self.limit_query_param:
try:
limit = int(request.query_params[self.limit_query_param])
if limit < 0:
raise ValueError()
# Enforce maximum page size, if defined
if settings.MAX_PAGE_SIZE:
if limit == 0:
return settings.MAX_PAGE_SIZE
else:
return min(limit, settings.MAX_PAGE_SIZE)
return limit
except (KeyError, ValueError):
pass
return self.default_limit
def get_next_link(self):
# Pagination has been disabled
if not self.limit:
return None
return super().get_next_link()
def get_previous_link(self):
# Pagination has been disabled
if not self.limit:
return None
return super().get_previous_link()

12
netbox/netbox/api/renderers.py Normal file
View File

@ -0,0 +1,12 @@
from rest_framework.renderers import BrowsableAPIRenderer
class FormlessBrowsableAPIRenderer(BrowsableAPIRenderer):
"""
Override the built-in BrowsableAPIRenderer to disable HTML forms.
"""
def show_form_for_method(self, *args, **kwargs):
return False
def get_filter_form(self, data, view, request):
return None

27
netbox/netbox/api/routers.py Normal file
View File

@ -0,0 +1,27 @@
from collections import OrderedDict
from rest_framework.routers import DefaultRouter
class OrderedDefaultRouter(DefaultRouter):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Extend the list view mappings to support the DELETE operation
self.routes[0].mapping.update({
'put': 'bulk_update',
'patch': 'bulk_partial_update',
'delete': 'bulk_destroy',
})
def get_api_root_view(self, api_urls=None):
"""
Wrap DRF's DefaultRouter to return an alphabetized list of endpoints.
"""
api_root_dict = OrderedDict()
list_name = self.routes[0].name
for prefix, viewset, basename in sorted(self.registry, key=lambda x: x[0]):
api_root_dict[prefix] = list_name.format(basename=basename)
return self.APIRootView.as_view(api_root_dict=api_root_dict)

91
netbox/netbox/api/serializers.py Normal file
View File

@ -0,0 +1,91 @@
from django.core.exceptions import FieldError, MultipleObjectsReturned, ObjectDoesNotExist
from django.db.models import ManyToManyField
from rest_framework import serializers
from rest_framework.exceptions import ValidationError
from utilities.utils import dict_to_filter_params
# TODO: We should probably take a fresh look at exactly what we're doing with this. There might be a more elegant
# way to enforce model validation on the serializer.
class ValidatedModelSerializer(serializers.ModelSerializer):
"""
Extends the built-in ModelSerializer to enforce calling clean() on the associated model during validation.
"""
def validate(self, data):
# Remove custom fields data and tags (if any) prior to model validation
attrs = data.copy()
attrs.pop('custom_fields', None)
attrs.pop('tags', None)
# Skip ManyToManyFields
for field in self.Meta.model._meta.get_fields():
if isinstance(field, ManyToManyField):
attrs.pop(field.name, None)
# Run clean() on an instance of the model
if self.instance is None:
instance = self.Meta.model(**attrs)
else:
instance = self.instance
for k, v in attrs.items():
setattr(instance, k, v)
instance.clean()
instance.validate_unique()
return data
class WritableNestedSerializer(serializers.ModelSerializer):
"""
Returns a nested representation of an object on read, but accepts only a primary key on write.
"""
def to_internal_value(self, data):
if data is None:
return None
# Dictionary of related object attributes
if isinstance(data, dict):
params = dict_to_filter_params(data)
queryset = self.Meta.model.objects
try:
return queryset.get(**params)
except ObjectDoesNotExist:
raise ValidationError(
"Related object not found using the provided attributes: {}".format(params)
)
except MultipleObjectsReturned:
raise ValidationError(
"Multiple objects match the provided attributes: {}".format(params)
)
except FieldError as e:
raise ValidationError(e)
# Integer PK of related object
if isinstance(data, int):
pk = data
else:
try:
# PK might have been mistakenly passed as a string
pk = int(data)
except (TypeError, ValueError):
raise ValidationError(
"Related objects must be referenced by numeric ID or by dictionary of attributes. Received an "
"unrecognized value: {}".format(data)
)
# Look up object by PK
queryset = self.Meta.model.objects
try:
return queryset.get(pk=int(data))
except ObjectDoesNotExist:
raise ValidationError(
"Related object not found using the provided numeric ID: {}".format(pk)
)
class BulkOperationSerializer(serializers.Serializer):
id = serializers.IntegerField()

220
netbox/netbox/api/views.py Normal file
View File

@ -0,0 +1,220 @@
import logging
from django.core.exceptions import ObjectDoesNotExist, PermissionDenied
from django.db import transaction
from django.db.models import ProtectedError
from rest_framework import mixins, status
from rest_framework.response import Response
from rest_framework.viewsets import GenericViewSet
from netbox.api import BulkOperationSerializer
from netbox.api.exceptions import SerializerNotFound
from utilities.api import get_serializer_for_model
HTTP_ACTIONS = {
'GET': 'view',
'OPTIONS': None,
'HEAD': 'view',
'POST': 'add',
'PUT': 'change',
'PATCH': 'change',
'DELETE': 'delete',
}
#
# Mixins
#
class BulkUpdateModelMixin:
"""
Support bulk modification of objects using the list endpoint for a model. Accepts a PATCH action with a list of one
or more JSON objects, each specifying the numeric ID of an object to be updated as well as the attributes to be set.
For example:
PATCH /api/dcim/sites/
[
{
"id": 123,
"name": "New name"
},
{
"id": 456,
"status": "planned"
}
]
"""
def bulk_update(self, request, *args, **kwargs):
partial = kwargs.pop('partial', False)
serializer = BulkOperationSerializer(data=request.data, many=True)
serializer.is_valid(raise_exception=True)
qs = self.get_queryset().filter(
pk__in=[o['id'] for o in serializer.data]
)
# Map update data by object ID
update_data = {
obj.pop('id'): obj for obj in request.data
}
self.perform_bulk_update(qs, update_data, partial=partial)
return Response(status=status.HTTP_200_OK)
def perform_bulk_update(self, objects, update_data, partial):
with transaction.atomic():
for obj in objects:
data = update_data.get(obj.id)
serializer = self.get_serializer(obj, data=data, partial=partial)
serializer.is_valid(raise_exception=True)
self.perform_update(serializer)
def bulk_partial_update(self, request, *args, **kwargs):
kwargs['partial'] = True
return self.bulk_update(request, *args, **kwargs)
class BulkDestroyModelMixin:
"""
Support bulk deletion of objects using the list endpoint for a model. Accepts a DELETE action with a list of one
or more JSON objects, each specifying the numeric ID of an object to be deleted. For example:
DELETE /api/dcim/sites/
[
{"id": 123},
{"id": 456}
]
"""
def bulk_destroy(self, request, *args, **kwargs):
serializer = BulkOperationSerializer(data=request.data, many=True)
serializer.is_valid(raise_exception=True)
qs = self.get_queryset().filter(
pk__in=[o['id'] for o in serializer.data]
)
self.perform_bulk_destroy(qs)
return Response(status=status.HTTP_204_NO_CONTENT)
def perform_bulk_destroy(self, objects):
with transaction.atomic():
for obj in objects:
self.perform_destroy(obj)
#
# Viewsets
#
class ModelViewSet(mixins.CreateModelMixin,
mixins.RetrieveModelMixin,
mixins.UpdateModelMixin,
mixins.DestroyModelMixin,
mixins.ListModelMixin,
BulkUpdateModelMixin,
BulkDestroyModelMixin,
GenericViewSet):
"""
Accept either a single object or a list of objects to create.
"""
def get_serializer(self, *args, **kwargs):
# If a list of objects has been provided, initialize the serializer with many=True
if isinstance(kwargs.get('data', {}), list):
kwargs['many'] = True
return super().get_serializer(*args, **kwargs)
def get_serializer_class(self):
logger = logging.getLogger('netbox.api.views.ModelViewSet')
# If 'brief' has been passed as a query param, find and return the nested serializer for this model, if one
# exists
request = self.get_serializer_context()['request']
if request.query_params.get('brief'):
logger.debug("Request is for 'brief' format; initializing nested serializer")
try:
serializer = get_serializer_for_model(self.queryset.model, prefix='Nested')
logger.debug(f"Using serializer {serializer}")
return serializer
except SerializerNotFound:
pass
# Fall back to the hard-coded serializer class
logger.debug(f"Using serializer {self.serializer_class}")
return self.serializer_class
def initial(self, request, *args, **kwargs):
super().initial(request, *args, **kwargs)
if not request.user.is_authenticated:
return
# Restrict the view's QuerySet to allow only the permitted objects
action = HTTP_ACTIONS[request.method]
if action:
self.queryset = self.queryset.restrict(request.user, action)
def dispatch(self, request, *args, **kwargs):
logger = logging.getLogger('netbox.api.views.ModelViewSet')
try:
return super().dispatch(request, *args, **kwargs)
except ProtectedError as e:
protected_objects = list(e.protected_objects)
msg = f'Unable to delete object. {len(protected_objects)} dependent objects were found: '
msg += ', '.join([f'{obj} ({obj.pk})' for obj in protected_objects])
logger.warning(msg)
return self.finalize_response(
request,
Response({'detail': msg}, status=409),
*args,
**kwargs
)
def _validate_objects(self, instance):
"""
Check that the provided instance or list of instances are matched by the current queryset. This confirms that
any newly created or modified objects abide by the attributes granted by any applicable ObjectPermissions.
"""
if type(instance) is list:
# Check that all instances are still included in the view's queryset
conforming_count = self.queryset.filter(pk__in=[obj.pk for obj in instance]).count()
if conforming_count != len(instance):
raise ObjectDoesNotExist
else:
# Check that the instance is matched by the view's queryset
self.queryset.get(pk=instance.pk)
def perform_create(self, serializer):
model = self.queryset.model
logger = logging.getLogger('netbox.api.views.ModelViewSet')
logger.info(f"Creating new {model._meta.verbose_name}")
# Enforce object-level permissions on save()
try:
with transaction.atomic():
instance = serializer.save()
self._validate_objects(instance)
except ObjectDoesNotExist:
raise PermissionDenied()
def perform_update(self, serializer):
model = self.queryset.model
logger = logging.getLogger('netbox.api.views.ModelViewSet')
logger.info(f"Updating {model._meta.verbose_name} {serializer.instance} (PK: {serializer.instance.pk})")
# Enforce object-level permissions on save()
try:
with transaction.atomic():
instance = serializer.save()
self._validate_objects(instance)
except ObjectDoesNotExist:
raise PermissionDenied()
def perform_destroy(self, instance):
model = self.queryset.model
logger = logging.getLogger('netbox.api.views.ModelViewSet')
logger.info(f"Deleting {model._meta.verbose_name} {instance} (PK: {instance.pk})")
return super().perform_destroy(instance)

10
netbox/netbox/settings.py
View File

@ -461,18 +461,18 @@ REST_FRAMEWORK = {
'ALLOWED_VERSIONS': [REST_FRAMEWORK_VERSION],
'DEFAULT_AUTHENTICATION_CLASSES': (
'rest_framework.authentication.SessionAuthentication',
'netbox.api.TokenAuthentication',
'netbox.api.authentication.TokenAuthentication',
),
'DEFAULT_FILTER_BACKENDS': (
'django_filters.rest_framework.DjangoFilterBackend',
),
'DEFAULT_PAGINATION_CLASS': 'netbox.api.OptionalLimitOffsetPagination',
'DEFAULT_PAGINATION_CLASS': 'netbox.api.pagination.OptionalLimitOffsetPagination',
'DEFAULT_PERMISSION_CLASSES': (
'netbox.api.TokenPermissions',
'netbox.api.authentication.TokenPermissions',
),
'DEFAULT_RENDERER_CLASSES': (
'rest_framework.renderers.JSONRenderer',
'netbox.api.FormlessBrowsableAPIRenderer',
'netbox.api.renderers.FormlessBrowsableAPIRenderer',
),
'DEFAULT_VERSION': REST_FRAMEWORK_VERSION,
'DEFAULT_VERSIONING_CLASS': 'rest_framework.versioning.AcceptHeaderVersioning',
@ -484,7 +484,7 @@ REST_FRAMEWORK = {
# Custom operations
'bulk_destroy': 'bulk_delete',
},
'VIEW_NAME_FUNCTION': 'netbox.api.get_view_name',
'VIEW_NAME_FUNCTION': 'utilities.api.get_view_name',
}

2
netbox/secrets/api/nested_serializers.py
View File

@ -1,7 +1,7 @@
from rest_framework import serializers
from netbox.api import WritableNestedSerializer
from secrets.models import Secret, SecretRole
from utilities.api import WritableNestedSerializer
__all__ = [
'NestedSecretRoleSerializer',

3
netbox/secrets/api/serializers.py
View File

@ -6,7 +6,8 @@ from extras.api.customfields import CustomFieldModelSerializer
from extras.api.serializers import TaggedObjectSerializer
from secrets.constants import SECRET_ASSIGNMENT_MODELS
from secrets.models import Secret, SecretRole
from utilities.api import ContentTypeField, ValidatedModelSerializer, get_serializer_for_model
from netbox.api import ContentTypeField, ValidatedModelSerializer
from utilities.api import get_serializer_for_model
from .nested_serializers import *

2
netbox/secrets/api/urls.py
View File

@ -1,4 +1,4 @@
from utilities.api import OrderedDefaultRouter
from netbox.api import OrderedDefaultRouter
from . import views

2
netbox/secrets/api/views.py
View File

@ -9,10 +9,10 @@ from rest_framework.response import Response
from rest_framework.routers import APIRootView
from rest_framework.viewsets import ViewSet
from netbox.api.views import ModelViewSet
from secrets import filters
from secrets.exceptions import InvalidKey
from secrets.models import Secret, SecretRole, SessionKey, UserKey
from utilities.api import ModelViewSet
from . import serializers
ERR_USERKEY_MISSING = "No UserKey found for the current user."

2
netbox/tenancy/api/nested_serializers.py
View File

@ -1,7 +1,7 @@
from rest_framework import serializers
from netbox.api import WritableNestedSerializer
from tenancy.models import Tenant, TenantGroup
from utilities.api import WritableNestedSerializer
__all__ = [
'NestedTenantGroupSerializer',

2
netbox/tenancy/api/serializers.py
View File

@ -2,8 +2,8 @@ from rest_framework import serializers
from extras.api.customfields import CustomFieldModelSerializer
from extras.api.serializers import TaggedObjectSerializer
from netbox.api import ValidatedModelSerializer
from tenancy.models import Tenant, TenantGroup
from utilities.api import ValidatedModelSerializer
from .nested_serializers import *

2
netbox/tenancy/api/urls.py
View File

@ -1,4 +1,4 @@
from utilities.api import OrderedDefaultRouter
from netbox.api import OrderedDefaultRouter
from . import views

2
netbox/tenancy/api/views.py
View File

@ -4,9 +4,9 @@ from circuits.models import Circuit
from dcim.models import Device, Rack, Site
from extras.api.views import CustomFieldModelViewSet
from ipam.models import IPAddress, Prefix, VLAN, VRF
from netbox.api.views import ModelViewSet
from tenancy import filters
from tenancy.models import Tenant, TenantGroup
from utilities.api import ModelViewSet
from utilities.utils import get_subquery
from virtualization.models import VirtualMachine
from . import serializers

2
netbox/users/api/nested_serializers.py
View File

@ -2,8 +2,8 @@ from django.contrib.auth.models import Group, User
from django.contrib.contenttypes.models import ContentType
from rest_framework import serializers
from netbox.api import ContentTypeField, WritableNestedSerializer
from users.models import ObjectPermission
from utilities.api import ContentTypeField, WritableNestedSerializer
__all__ = [
'NestedGroupSerializer',

2
netbox/users/api/serializers.py
View File

@ -2,8 +2,8 @@ from django.contrib.auth.models import Group, User
from django.contrib.contenttypes.models import ContentType
from rest_framework import serializers
from netbox.api import ContentTypeField, SerializedPKRelatedField, ValidatedModelSerializer
from users.models import ObjectPermission
from utilities.api import ContentTypeField, SerializedPKRelatedField, ValidatedModelSerializer
from .nested_serializers import *

2
netbox/users/api/urls.py
View File

@ -1,4 +1,4 @@
from utilities.api import OrderedDefaultRouter
from netbox.api import OrderedDefaultRouter
from . import views

2
netbox/users/api/views.py
View File

@ -2,9 +2,9 @@ from django.contrib.auth.models import Group, User
from django.db.models import Count
from rest_framework.routers import APIRootView
from netbox.api.views import ModelViewSet
from users import filters
from users.models import ObjectPermission
from utilities.api import ModelViewSet
from utilities.querysets import RestrictedQuerySet
from . import serializers

511
netbox/utilities/api.py
View File

@ -1,41 +1,8 @@
import logging
from collections import OrderedDict
import pytz
from django.conf import settings
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import FieldError, MultipleObjectsReturned, ObjectDoesNotExist, PermissionDenied
from django.db import transaction
from django.db.models import ManyToManyField, ProtectedError
from django.urls import reverse
from rest_framework import mixins, serializers, status
from rest_framework.exceptions import APIException, ValidationError
from rest_framework.permissions import BasePermission
from rest_framework.relations import PrimaryKeyRelatedField, RelatedField
from rest_framework.response import Response
from rest_framework.routers import DefaultRouter
from rest_framework.viewsets import GenericViewSet
from rest_framework.utils import formatting
from .utils import dict_to_filter_params, dynamic_import
HTTP_ACTIONS = {
'GET': 'view',
'OPTIONS': None,
'HEAD': 'view',
'POST': 'add',
'PUT': 'change',
'PATCH': 'change',
'DELETE': 'delete',
}
class ServiceUnavailable(APIException):
status_code = 503
default_detail = "Service temporarily unavailable, please try again later."
class SerializerNotFound(Exception):
pass
from netbox.api.exceptions import SerializerNotFound
from .utils import dynamic_import
def get_serializer_for_model(model, prefix=''):
@ -63,459 +30,23 @@ def is_api_request(request):
return request.path_info.startswith(api_path)
#
# Authentication
#
class IsAuthenticatedOrLoginNotRequired(BasePermission):
def get_view_name(view, suffix=None):
"""
Returns True if the user is authenticated or LOGIN_REQUIRED is False.
Derive the view name from its associated model, if it has one. Fall back to DRF's built-in `get_view_name`.
"""
def has_permission(self, request, view):
if not settings.LOGIN_REQUIRED:
return True
return request.user.is_authenticated
#
# Fields
#
class ChoiceField(serializers.Field):
"""
Represent a ChoiceField as {'value': <DB value>, 'label': <string>}. Accepts a single value on write.
:param choices: An iterable of choices in the form (value, key).
:param allow_blank: Allow blank values in addition to the listed choices.
"""
def __init__(self, choices, allow_blank=False, **kwargs):
self.choiceset = choices
self.allow_blank = allow_blank
self._choices = dict()
# Unpack grouped choices
for k, v in choices:
if type(v) in [list, tuple]:
for k2, v2 in v:
self._choices[k2] = v2
else:
self._choices[k] = v
super().__init__(**kwargs)
def validate_empty_values(self, data):
# Convert null to an empty string unless allow_null == True
if data is None:
if self.allow_null:
return True, None
else:
data = ''
return super().validate_empty_values(data)
def to_representation(self, obj):
if obj is '':
return None
return OrderedDict([
('value', obj),
('label', self._choices[obj])
])
def to_internal_value(self, data):
if data is '':
if self.allow_blank:
return data
raise ValidationError("This field may not be blank.")
# Provide an explicit error message if the request is trying to write a dict or list
if isinstance(data, (dict, list)):
raise ValidationError('Value must be passed directly (e.g. "foo": 123); do not use a dictionary or list.')
# Check for string representations of boolean/integer values
if hasattr(data, 'lower'):
if data.lower() == 'true':
data = True
elif data.lower() == 'false':
data = False
else:
try:
data = int(data)
except ValueError:
pass
try:
if data in self._choices:
return data
except TypeError: # Input is an unhashable type
pass
raise ValidationError(f"{data} is not a valid choice.")
@property
def choices(self):
return self._choices
class ContentTypeField(RelatedField):
"""
Represent a ContentType as '<app_label>.<model>'
"""
default_error_messages = {
"does_not_exist": "Invalid content type: {content_type}",
"invalid": "Invalid value. Specify a content type as '<app_label>.<model_name>'.",
}
def to_internal_value(self, data):
try:
app_label, model = data.split('.')
return ContentType.objects.get_by_natural_key(app_label=app_label, model=model)
except ObjectDoesNotExist:
self.fail('does_not_exist', content_type=data)
except (TypeError, ValueError):
self.fail('invalid')
def to_representation(self, obj):
return "{}.{}".format(obj.app_label, obj.model)
class TimeZoneField(serializers.Field):
"""
Represent a pytz time zone.
"""
def to_representation(self, obj):
return obj.zone if obj else None
def to_internal_value(self, data):
if not data:
return ""
if data not in pytz.common_timezones:
raise ValidationError('Unknown time zone "{}" (see pytz.common_timezones for all options)'.format(data))
return pytz.timezone(data)
class SerializedPKRelatedField(PrimaryKeyRelatedField):
"""
Extends PrimaryKeyRelatedField to return a serialized object on read. This is useful for representing related
objects in a ManyToManyField while still allowing a set of primary keys to be written.
"""
def __init__(self, serializer, **kwargs):
self.serializer = serializer
self.pk_field = kwargs.pop('pk_field', None)
super().__init__(**kwargs)
def to_representation(self, value):
return self.serializer(value, context={'request': self.context['request']}).data
#
# Serializers
#
# TODO: We should probably take a fresh look at exactly what we're doing with this. There might be a more elegant
# way to enforce model validation on the serializer.
class ValidatedModelSerializer(serializers.ModelSerializer):
"""
Extends the built-in ModelSerializer to enforce calling clean() on the associated model during validation.
"""
def validate(self, data):
# Remove custom fields data and tags (if any) prior to model validation
attrs = data.copy()
attrs.pop('custom_fields', None)
attrs.pop('tags', None)
# Skip ManyToManyFields
for field in self.Meta.model._meta.get_fields():
if isinstance(field, ManyToManyField):
attrs.pop(field.name, None)
# Run clean() on an instance of the model
if self.instance is None:
instance = self.Meta.model(**attrs)
else:
instance = self.instance
for k, v in attrs.items():
setattr(instance, k, v)
instance.clean()
instance.validate_unique()
return data
class WritableNestedSerializer(serializers.ModelSerializer):
"""
Returns a nested representation of an object on read, but accepts only a primary key on write.
"""
def to_internal_value(self, data):
if data is None:
return None
# Dictionary of related object attributes
if isinstance(data, dict):
params = dict_to_filter_params(data)
queryset = self.Meta.model.objects
try:
return queryset.get(**params)
except ObjectDoesNotExist:
raise ValidationError(
"Related object not found using the provided attributes: {}".format(params)
)
except MultipleObjectsReturned:
raise ValidationError(
"Multiple objects match the provided attributes: {}".format(params)
)
except FieldError as e:
raise ValidationError(e)
# Integer PK of related object
if isinstance(data, int):
pk = data
else:
try:
# PK might have been mistakenly passed as a string
pk = int(data)
except (TypeError, ValueError):
raise ValidationError(
"Related objects must be referenced by numeric ID or by dictionary of attributes. Received an "
"unrecognized value: {}".format(data)
)
# Look up object by PK
queryset = self.Meta.model.objects
try:
return queryset.get(pk=int(data))
except ObjectDoesNotExist:
raise ValidationError(
"Related object not found using the provided numeric ID: {}".format(pk)
)
class BulkOperationSerializer(serializers.Serializer):
id = serializers.IntegerField()
#
# Mixins
#
class BulkUpdateModelMixin:
"""
Support bulk modification of objects using the list endpoint for a model. Accepts a PATCH action with a list of one
or more JSON objects, each specifying the numeric ID of an object to be updated as well as the attributes to be set.
For example:
PATCH /api/dcim/sites/
[
{
"id": 123,
"name": "New name"
},
{
"id": 456,
"status": "planned"
}
]
"""
def bulk_update(self, request, *args, **kwargs):
partial = kwargs.pop('partial', False)
serializer = BulkOperationSerializer(data=request.data, many=True)
serializer.is_valid(raise_exception=True)
qs = self.get_queryset().filter(
pk__in=[o['id'] for o in serializer.data]
)
# Map update data by object ID
update_data = {
obj.pop('id'): obj for obj in request.data
}
self.perform_bulk_update(qs, update_data, partial=partial)
return Response(status=status.HTTP_200_OK)
def perform_bulk_update(self, objects, update_data, partial):
with transaction.atomic():
for obj in objects:
data = update_data.get(obj.id)
serializer = self.get_serializer(obj, data=data, partial=partial)
serializer.is_valid(raise_exception=True)
self.perform_update(serializer)
def bulk_partial_update(self, request, *args, **kwargs):
kwargs['partial'] = True
return self.bulk_update(request, *args, **kwargs)
class BulkDestroyModelMixin:
"""
Support bulk deletion of objects using the list endpoint for a model. Accepts a DELETE action with a list of one
or more JSON objects, each specifying the numeric ID of an object to be deleted. For example:
DELETE /api/dcim/sites/
[
{"id": 123},
{"id": 456}
]
"""
def bulk_destroy(self, request, *args, **kwargs):
serializer = BulkOperationSerializer(data=request.data, many=True)
serializer.is_valid(raise_exception=True)
qs = self.get_queryset().filter(
pk__in=[o['id'] for o in serializer.data]
)
self.perform_bulk_destroy(qs)
return Response(status=status.HTTP_204_NO_CONTENT)
def perform_bulk_destroy(self, objects):
with transaction.atomic():
for obj in objects:
self.perform_destroy(obj)
#
# Viewsets
#
class ModelViewSet(mixins.CreateModelMixin,
mixins.RetrieveModelMixin,
mixins.UpdateModelMixin,
mixins.DestroyModelMixin,
mixins.ListModelMixin,
BulkUpdateModelMixin,
BulkDestroyModelMixin,
GenericViewSet):
"""
Accept either a single object or a list of objects to create.
"""
def get_serializer(self, *args, **kwargs):
# If a list of objects has been provided, initialize the serializer with many=True
if isinstance(kwargs.get('data', {}), list):
kwargs['many'] = True
return super().get_serializer(*args, **kwargs)
def get_serializer_class(self):
logger = logging.getLogger('netbox.api.views.ModelViewSet')
# If 'brief' has been passed as a query param, find and return the nested serializer for this model, if one
# exists
request = self.get_serializer_context()['request']
if request.query_params.get('brief'):
logger.debug("Request is for 'brief' format; initializing nested serializer")
try:
serializer = get_serializer_for_model(self.queryset.model, prefix='Nested')
logger.debug(f"Using serializer {serializer}")
return serializer
except SerializerNotFound:
pass
# Fall back to the hard-coded serializer class
logger.debug(f"Using serializer {self.serializer_class}")
return self.serializer_class
def initial(self, request, *args, **kwargs):
super().initial(request, *args, **kwargs)
if not request.user.is_authenticated:
return
# Restrict the view's QuerySet to allow only the permitted objects
action = HTTP_ACTIONS[request.method]
if action:
self.queryset = self.queryset.restrict(request.user, action)
def dispatch(self, request, *args, **kwargs):
logger = logging.getLogger('netbox.api.views.ModelViewSet')
try:
return super().dispatch(request, *args, **kwargs)
except ProtectedError as e:
protected_objects = list(e.protected_objects)
msg = f'Unable to delete object. {len(protected_objects)} dependent objects were found: '
msg += ', '.join([f'{obj} ({obj.pk})' for obj in protected_objects])
logger.warning(msg)
return self.finalize_response(
request,
Response({'detail': msg}, status=409),
*args,
**kwargs
)
def _validate_objects(self, instance):
"""
Check that the provided instance or list of instances are matched by the current queryset. This confirms that
any newly created or modified objects abide by the attributes granted by any applicable ObjectPermissions.
"""
if type(instance) is list:
# Check that all instances are still included in the view's queryset
conforming_count = self.queryset.filter(pk__in=[obj.pk for obj in instance]).count()
if conforming_count != len(instance):
raise ObjectDoesNotExist
else:
# Check that the instance is matched by the view's queryset
self.queryset.get(pk=instance.pk)
def perform_create(self, serializer):
model = self.queryset.model
logger = logging.getLogger('netbox.api.views.ModelViewSet')
logger.info(f"Creating new {model._meta.verbose_name}")
# Enforce object-level permissions on save()
try:
with transaction.atomic():
instance = serializer.save()
self._validate_objects(instance)
except ObjectDoesNotExist:
raise PermissionDenied()
def perform_update(self, serializer):
model = self.queryset.model
logger = logging.getLogger('netbox.api.views.ModelViewSet')
logger.info(f"Updating {model._meta.verbose_name} {serializer.instance} (PK: {serializer.instance.pk})")
# Enforce object-level permissions on save()
try:
with transaction.atomic():
instance = serializer.save()
self._validate_objects(instance)
except ObjectDoesNotExist:
raise PermissionDenied()
def perform_destroy(self, instance):
model = self.queryset.model
logger = logging.getLogger('netbox.api.views.ModelViewSet')
logger.info(f"Deleting {model._meta.verbose_name} {instance} (PK: {instance.pk})")
return super().perform_destroy(instance)
#
# Routers
#
class OrderedDefaultRouter(DefaultRouter):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Extend the list view mappings to support the DELETE operation
self.routes[0].mapping.update({
'put': 'bulk_update',
'patch': 'bulk_partial_update',
'delete': 'bulk_destroy',
})
def get_api_root_view(self, api_urls=None):
"""
Wrap DRF's DefaultRouter to return an alphabetized list of endpoints.
"""
api_root_dict = OrderedDict()
list_name = self.routes[0].name
for prefix, viewset, basename in sorted(self.registry, key=lambda x: x[0]):
api_root_dict[prefix] = list_name.format(basename=basename)
return self.APIRootView.as_view(api_root_dict=api_root_dict)
if hasattr(view, 'queryset'):
# Determine the model name from the queryset.
name = view.queryset.model._meta.verbose_name
name = ' '.join([w[0].upper() + w[1:] for w in name.split()]) # Capitalize each word
else:
# Replicate DRF's built-in behavior.
name = view.__class__.__name__
name = formatting.remove_trailing_string(name, 'View')
name = formatting.remove_trailing_string(name, 'ViewSet')
name = formatting.camelcase_to_spaces(name)
if suffix:
name += ' ' + suffix
return name

2
netbox/utilities/custom_inspectors.py
View File

@ -6,7 +6,7 @@ from rest_framework.fields import ChoiceField
from rest_framework.relations import ManyRelatedField
from extras.api.customfields import CustomFieldsDataField
from utilities.api import ChoiceField, SerializedPKRelatedField, WritableNestedSerializer
from netbox.api import ChoiceField, SerializedPKRelatedField, WritableNestedSerializer
class NetBoxSwaggerAutoSchema(SwaggerAutoSchema):

2
netbox/virtualization/api/nested_serializers.py
View File

@ -1,7 +1,7 @@
from rest_framework import serializers
from dcim.models import Interface
from utilities.api import WritableNestedSerializer
from netbox.api import WritableNestedSerializer
from virtualization.models import Cluster, ClusterGroup, ClusterType, VirtualMachine
__all__ = [

2
netbox/virtualization/api/serializers.py
View File

@ -7,8 +7,8 @@ from extras.api.customfields import CustomFieldModelSerializer
from extras.api.serializers import TaggedObjectSerializer
from ipam.api.nested_serializers import NestedIPAddressSerializer, NestedVLANSerializer
from ipam.models import VLAN
from netbox.api import ChoiceField, SerializedPKRelatedField, ValidatedModelSerializer
from tenancy.api.nested_serializers import NestedTenantSerializer
from utilities.api import ChoiceField, SerializedPKRelatedField, ValidatedModelSerializer
from virtualization.choices import *
from virtualization.models import Cluster, ClusterGroup, ClusterType, VirtualMachine, VMInterface
from .nested_serializers import *

2
netbox/virtualization/api/urls.py
View File

@ -1,4 +1,4 @@
from utilities.api import OrderedDefaultRouter
from netbox.api import OrderedDefaultRouter
from . import views

2
netbox/virtualization/api/views.py
View File

@ -3,7 +3,7 @@ from rest_framework.routers import APIRootView
from dcim.models import Device
from extras.api.views import CustomFieldModelViewSet
from utilities.api import ModelViewSet
from netbox.api.views import ModelViewSet
from utilities.utils import get_subquery
from virtualization import filters
from virtualization.models import Cluster, ClusterGroup, ClusterType, VirtualMachine, VMInterface