def determine_version(self, request, *args, **kwargs):
resolver_match = getattr(request, 'resolver_match', None)
if resolver_match is None or not resolver_match.namespace:
return self.default_version
# Allow for possibly nested namespaces.
possible_versions = resolver_match.namespace.split(':')
for version in possible_versions:
if self.is_allowed_version(version):
return version
raise exceptions.NotFound(self.invalid_version_message)
python类NotFound()的实例源码
def determine_version(self, request, *args, **kwargs):
hostname, separator, port = request.get_host().partition(':')
match = self.hostname_regex.match(hostname)
if not match:
return self.default_version
version = match.group(1)
if not self.is_allowed_version(version):
raise exceptions.NotFound(self.invalid_version_message)
return version
# We don't need to implement `reverse`, as the hostname will already be
# preserved as part of the REST framework `reverse` implementation.
def determine_version(self, request, *args, **kwargs):
version = request.query_params.get(self.version_param, self.default_version)
if not self.is_allowed_version(version):
raise exceptions.NotFound(self.invalid_version_message)
return version
def paginate_queryset(self, queryset, request, view=None):
"""
Paginate a queryset if required, either returning a
page object, or `None` if pagination is not configured for this view.
"""
page_size = self.get_page_size(request)
if not page_size:
return None
paginator = self.django_paginator_class(queryset, page_size)
page_number = request.query_params.get(self.page_query_param, 1)
if page_number in self.last_page_strings:
page_number = paginator.num_pages
try:
self.page = paginator.page(page_number)
except InvalidPage as exc:
msg = self.invalid_page_message.format(
page_number=page_number, message=six.text_type(exc)
)
raise NotFound(msg)
if paginator.num_pages > 1 and self.template is not None:
# The browsable API should display pagination controls.
self.display_page_controls = True
self.request = request
return list(self.page)
def decode_cursor(self, request):
"""
Given a request with a cursor, return a `Cursor` instance.
Differs from the standard CursorPagination to handle a tuple in the
position field.
"""
# Determine if we have a cursor, and if so then decode it.
encoded = request.query_params.get(self.cursor_query_param)
if encoded is None:
return None
try:
querystring = b64decode(encoded.encode('ascii')).decode('ascii')
tokens = urlparse.parse_qs(querystring, keep_blank_values=True)
offset = tokens.get('o', ['0'])[0]
# This was hard-coded until Django REST Framework 3.4.0.
try:
offset_cutoff = self.offset_cutoff
except AttributeError:
offset_cutoff = 1000
offset = _positive_int(offset, cutoff=offset_cutoff)
reverse = tokens.get('r', ['0'])[0]
reverse = bool(int(reverse))
# The difference. Don't get just the 0th entry: get all entries.
position = tokens.get('p', None)
except (TypeError, ValueError):
raise NotFound(self.invalid_cursor_message)
return Cursor(offset=offset, reverse=reverse, position=position)
def test_invalid_cursor(self):
request = Request(factory.get('/', {'cursor': '123'}))
with self.assertRaises(exceptions.NotFound):
self.pagination.paginate_queryset(self.queryset, request)
def get(self, request, *args, **kwargs):
company = request.user.company
notification_id = kwargs['notification_id']
try:
note = Notification.objects.get(company=company, id=notification_id)
except Notification.DoesNotExist:
raise exceptions.NotFound()
serializer = self.get_serializer(note)
return Response({'status': 'success', 'data': serializer.data})
def patch(self, request, *args, **kwargs):
company = request.user.company
notification_id = kwargs['notification_id']
try:
note = Notification.objects.get(company=company, id=notification_id)
except Notification.DoesNotExist:
raise exceptions.NotFound()
serializer = self.get_serializer(note, data=request.data, partial=True)
serializer.is_valid(raise_exception=True)
instance = serializer.save()
return Response({'status': 'success', 'data': serializer.data})
def delete(self, request, *args, **kwargs):
company = request.user.company
notification_id = kwargs['notification_id']
try:
note = Notification.objects.get(company=company, id=notification_id)
except Notification.DoesNotExist:
raise exceptions.NotFound()
serializer = self.get_serializer(note)
instance = serializer.delete()
return Response({'status': 'success'})
def get(self, request, *args, **kwargs):
company = request.user.company
log_id = kwargs['log_id']
try:
log = NotificationLog.objects.get(notification__company=company,
id=log_id)
except NotificationLog.DoesNotExist:
raise exceptions.NotFound()
serializer = self.get_serializer(log)
return Response({'status': 'success', 'data': serializer.data})
def get(self, request, *args, **kwargs):
company = request.user.company
notification_id = kwargs['notification_id']
try:
note = Notification.objects.get(company=company, id=notification_id,
preference_enabled=True)
except Notification.DoesNotExist:
raise exceptions.NotFound()
serializer = self.get_serializer(note)
return Response({'status': 'success', 'data': serializer.data})
def get_last_build(request, package_name, version=None):
account = request.user
package = get_object_or_404(Package, name=package_name, account=account)
last_build = package.builds.filter(version=version).order_by('-build_number').first()
if not last_build:
raise exceptions.NotFound("Couldn't find a build for this package name")
# Serialize the new build
serializer = BuildSerializer(last_build)
response_data = serializer.data.copy()
response_data.update(last_build.extra)
return Response(response_data)
def determine_version(self, request, *args, **kwargs):
version = kwargs.get(self.version_param, self.default_version)
if not self.is_allowed_version(version):
raise exceptions.NotFound(self.invalid_version_message)
return version
def determine_version(self, request, *args, **kwargs):
resolver_match = getattr(request, 'resolver_match', None)
if (resolver_match is None or not resolver_match.namespace):
return self.default_version
version = resolver_match.namespace
if not self.is_allowed_version(version):
raise exceptions.NotFound(self.invalid_version_message)
return version
def determine_version(self, request, *args, **kwargs):
hostname, seperator, port = request.get_host().partition(':')
match = self.hostname_regex.match(hostname)
if not match:
return self.default_version
version = match.group(1)
if not self.is_allowed_version(version):
raise exceptions.NotFound(self.invalid_version_message)
return version
# We don't need to implement `reverse`, as the hostname will already be
# preserved as part of the REST framework `reverse` implementation.
def determine_version(self, request, *args, **kwargs):
version = request.query_params.get(self.version_param, self.default_version)
if not self.is_allowed_version(version):
raise exceptions.NotFound(self.invalid_version_message)
return version
def paginate_queryset(self, queryset, request, view=None):
"""
Paginate a queryset if required, either returning a
page object, or `None` if pagination is not configured for this view.
"""
page_size = self.get_page_size(request)
if not page_size:
return None
paginator = self.django_paginator_class(queryset, page_size)
page_number = request.query_params.get(self.page_query_param, 1)
if page_number in self.last_page_strings:
page_number = paginator.num_pages
try:
self.page = paginator.page(page_number)
except InvalidPage as exc:
msg = self.invalid_page_message.format(
page_number=page_number, message=six.text_type(exc)
)
raise NotFound(msg)
if paginator.num_pages > 1 and self.template is not None:
# The browsable API should display pagination controls.
self.display_page_controls = True
self.request = request
return list(self.page)
def determine_version(self, request, *args, **kwargs):
resolver_match = getattr(request, 'resolver_match', None)
if (resolver_match is None or not resolver_match.namespace):
return self.default_version
version = resolver_match.namespace.split(':')[0]
if not self.is_allowed_version(version):
raise exceptions.NotFound(self.invalid_version_message)
return version
def get_queryset(self):
movie_pk = self.kwargs['pk']
result = Movie.objects.filter(pk=movie_pk)
if result:
return result
raise NotFound('?? ??? ????.')
def metrics(request):
metrics = Metric.objects.all().order_by('name')
ds = request.query_params.get('ds')
if ds:
# Further filter metrics by dataset.
metrics = (metrics.filter(collection__dataset__slug=ds)
.distinct('id', 'name'))
if not metrics:
raise NotFound('No data set with given dataset found.')
return Response([MetricSerializer(m).data for m in metrics])
