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

Moved core API classes out of utilities

This commit is contained in:
Jeremy Stretch
2017-10-10 14:21:20 -04:00
parent 8403e91fc9
commit afbbe1148f
3 changed files with 145 additions and 135 deletions

139
netbox/netbox/api.py Normal file
View File

@ -0,0 +1,139 @@
from __future__ import unicode_literals
from rest_framework import authentication, exceptions
from rest_framework.pagination import LimitOffsetPagination
from rest_framework.permissions import DjangoModelPermissions, SAFE_METHODS
from rest_framework.renderers import BrowsableAPIRenderer
from rest_framework.views import get_view_name as drf_get_view_name
from users.models import Token
#
# Renderers
#
class FormlessBrowsableAPIRenderer(BrowsableAPIRenderer):
"""
Override the built-in BrowsableAPIRenderer to disable HTML forms.
"""
def show_form_for_method(self, *args, **kwargs):
return False
#
# 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.select_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(DjangoModelPermissions):
"""
Custom permissions handler which extends the built-in DjangoModelPermissions to validate a Token's write ability
for unsafe requests (POST/PUT/PATCH/DELETE).
"""
def __init__(self):
# LOGIN_REQUIRED determines whether read-only access is provided to anonymous users.
from django.conf import settings
self.authenticated_users_only = settings.LOGIN_REQUIRED
super(TokenPermissions, self).__init__()
def has_permission(self, request, view):
# If token authentication is in use, verify that the token allows write operations (for unsafe methods).
if request.method not in SAFE_METHODS and isinstance(request.auth, Token):
if not request.auth.write_enabled:
return False
return super(TokenPermissions, self).has_permission(request, view)
#
# 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):
try:
self.count = queryset.count()
except (AttributeError, TypeError):
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):
from django.conf import settings
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
#
# Miscellaneous
#
def get_view_name(view_cls, 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_cls, 'queryset'):
name = view_cls.queryset.model._meta.verbose_name
name = ' '.join([w[0].upper() + w[1:] for w in name.split()]) # Capitalize each word
if suffix:
name = "{} {}".format(name, suffix)
return name
return drf_get_view_name(view_cls, suffix)

View File

@ -211,23 +211,23 @@ REST_FRAMEWORK = {
'ALLOWED_VERSIONS': [REST_FRAMEWORK_VERSION], 'ALLOWED_VERSIONS': [REST_FRAMEWORK_VERSION],
'DEFAULT_AUTHENTICATION_CLASSES': ( 'DEFAULT_AUTHENTICATION_CLASSES': (
'rest_framework.authentication.SessionAuthentication', 'rest_framework.authentication.SessionAuthentication',
'utilities.api.TokenAuthentication', 'netbox.api.TokenAuthentication',
), ),
'DEFAULT_FILTER_BACKENDS': ( 'DEFAULT_FILTER_BACKENDS': (
'rest_framework.filters.DjangoFilterBackend', 'rest_framework.filters.DjangoFilterBackend',
), ),
'DEFAULT_PAGINATION_CLASS': 'utilities.api.OptionalLimitOffsetPagination', 'DEFAULT_PAGINATION_CLASS': 'netbox.api.OptionalLimitOffsetPagination',
'DEFAULT_PERMISSION_CLASSES': ( 'DEFAULT_PERMISSION_CLASSES': (
'utilities.api.TokenPermissions', 'netbox.api.TokenPermissions',
), ),
'DEFAULT_RENDERER_CLASSES': ( 'DEFAULT_RENDERER_CLASSES': (
'rest_framework.renderers.JSONRenderer', 'rest_framework.renderers.JSONRenderer',
'utilities.api.FormlessBrowsableAPIRenderer', 'netbox.api.FormlessBrowsableAPIRenderer',
), ),
'DEFAULT_VERSION': REST_FRAMEWORK_VERSION, 'DEFAULT_VERSION': REST_FRAMEWORK_VERSION,
'DEFAULT_VERSIONING_CLASS': 'rest_framework.versioning.AcceptHeaderVersioning', 'DEFAULT_VERSIONING_CLASS': 'rest_framework.versioning.AcceptHeaderVersioning',
'PAGE_SIZE': PAGINATE_COUNT, 'PAGE_SIZE': PAGINATE_COUNT,
'VIEW_NAME_FUNCTION': 'utilities.api.get_view_name', 'VIEW_NAME_FUNCTION': 'netbox.api.get_view_name',
} }
# Django debug toolbar # Django debug toolbar

View File

@ -3,16 +3,10 @@ from __future__ import unicode_literals
from django.conf import settings from django.conf import settings
from django.contrib.contenttypes.models import ContentType from django.contrib.contenttypes.models import ContentType
from rest_framework import authentication, exceptions
from rest_framework.compat import is_authenticated from rest_framework.compat import is_authenticated
from rest_framework.exceptions import APIException from rest_framework.exceptions import APIException
from rest_framework.pagination import LimitOffsetPagination from rest_framework.permissions import BasePermission
from rest_framework.permissions import BasePermission, DjangoModelPermissions, SAFE_METHODS
from rest_framework.renderers import BrowsableAPIRenderer
from rest_framework.serializers import Field, ModelSerializer, ValidationError from rest_framework.serializers import Field, ModelSerializer, ValidationError
from rest_framework.views import get_view_name as drf_get_view_name
from users.models import Token
WRITE_OPERATIONS = ['create', 'update', 'partial_update', 'delete'] WRITE_OPERATIONS = ['create', 'update', 'partial_update', 'delete']
@ -27,47 +21,6 @@ class ServiceUnavailable(APIException):
# Authentication # 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.select_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(DjangoModelPermissions):
"""
Custom permissions handler which extends the built-in DjangoModelPermissions to validate a Token's write ability
for unsafe requests (POST/PUT/PATCH/DELETE).
"""
def __init__(self):
# LOGIN_REQUIRED determines whether read-only access is provided to anonymous users.
self.authenticated_users_only = settings.LOGIN_REQUIRED
super(TokenPermissions, self).__init__()
def has_permission(self, request, view):
# If token authentication is in use, verify that the token allows write operations (for unsafe methods).
if request.method not in SAFE_METHODS and isinstance(request.auth, Token):
if not request.auth.write_enabled:
return False
return super(TokenPermissions, self).has_permission(request, view)
class IsAuthenticatedOrLoginNotRequired(BasePermission): class IsAuthenticatedOrLoginNotRequired(BasePermission):
""" """
Returns True if the user is authenticated or LOGIN_REQUIRED is False. Returns True if the user is authenticated or LOGIN_REQUIRED is False.
@ -153,85 +106,3 @@ class WritableSerializerMixin(object):
if self.action in WRITE_OPERATIONS and hasattr(self, 'write_serializer_class'): if self.action in WRITE_OPERATIONS and hasattr(self, 'write_serializer_class'):
return self.write_serializer_class return self.write_serializer_class
return self.serializer_class return self.serializer_class
#
# 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):
try:
self.count = queryset.count()
except (AttributeError, TypeError):
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
#
# Renderers
#
class FormlessBrowsableAPIRenderer(BrowsableAPIRenderer):
"""
Override the built-in BrowsableAPIRenderer to disable HTML forms.
"""
def show_form_for_method(self, *args, **kwargs):
return False
#
# Miscellaneous
#
def get_view_name(view_cls, 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_cls, 'queryset'):
name = view_cls.queryset.model._meta.verbose_name
name = ' '.join([w[0].upper() + w[1:] for w in name.split()]) # Capitalize each word
if suffix:
name = "{} {}".format(name, suffix)
return name
return drf_get_view_name(view_cls, suffix)