Python rest_framework.exceptions 模块，APIException() 实例源码
我们从Python开源项目中，提取了以下33个代码示例，用于说明如何使用rest_framework.exceptions.APIException()。
def run_action(self, action, pk, data):
try:
if not self.has_permission(self.user, action, pk):
self.reply(action, errors=['Permission Denied'], status=401)
if not action in self.available_actions:
self.reply(action, errors=['Invalid Action'], status=400)
elif action in ('create', 'list'):
data, status = getattr(self, action)(data)
elif action in ('retrieve', 'delete'):
data, status = getattr(self, action)(pk)
elif action in ('update', 'subscribe'):
data, status = getattr(self, action)(pk, data)
self.reply(action, data=data, status=status,
request_id=self.request_id)
except APIException as ex:
self.reply(action, errors=self._format_errors(ex.detail),
status=ex.status_code, request_id=self.request_id)
def friendly_exception_handler(exc, context):
response = exception_handler(exc, context)
if not response and settings.CATCH_ALL_EXCEPTIONS:
exc = APIException(exc)
response = exception_handler(exc, context)
if response is not None:
if is_pretty(response):
return response
error_message = response.data['detail']
error_code = settings.FRIENDLY_EXCEPTION_DICT.get(
exc.__class__.__name__)
response.data.pop('detail', {})
response.data['code'] = error_code
response.data['message'] = error_message
response.data['status_code'] = response.status_code
# response.data['exception'] = exc.__class__.__name__
return response
def run_action(self, action, pk, data):
try:
if not self.has_permission(self.user, action, pk):
self.reply(action, errors=['Permission Denied'], status=401,
request_id=self.request_id)
elif action not in self.available_actions:
self.reply(action, errors=['Invalid Action'], status=400,
request_id=self.request_id)
else:
methodname = self.available_actions[action]
method = getattr(self, methodname)
detail = getattr(method, 'detail', True)
if detail:
rv = method(pk, data=data)
else:
rv = method(data=data)
data, status = rv
self.reply(action, data=data, status=status, request_id=self.request_id)
except APIException as ex:
self.reply(action, errors=self._format_errors(ex.detail), status=ex.status_code, request_id=self.request_id)
def _authenticate(self):
"""
Attempt to authenticate the request using each authentication instance
in turn.
Returns a three-tuple of (authenticator, user, authtoken).
"""
for authenticator in self.authenticators:
try:
user_auth_tuple = authenticator.authenticate(self)
except exceptions.APIException:
self._not_authenticated()
raise
if user_auth_tuple is not None:
self._authenticator = authenticator
self.user, self.auth = user_auth_tuple
return
self._not_authenticated()
def _authenticate(self):
"""
Attempt to authenticate the request using each authentication instance
in turn.
Returns a three-tuple of (authenticator, user, authtoken).
"""
for authenticator in self.authenticators:
try:
user_auth_tuple = authenticator.authenticate(self)
except exceptions.APIException:
self._not_authenticated()
raise
if user_auth_tuple is not None:
self._authenticator = authenticator
self.user, self.auth = user_auth_tuple
return
self._not_authenticated()
def web_sight_exception_handler(exc, context):
"""
This is a custom Django exception handler that handles all exceptions thrown by the Web Sight REST
API.
:param exc: The exception that was thrown.
:param context: The context in which the exception was thrown.
:return: A response to return to the requesting user.
"""
if isinstance(exc, IntegrityError):
return handle_integrity_error(exc, context)
elif isinstance(exc, ValidationError):
return handle_validation_error(exc, context)
elif isinstance(exc, WsRestNonFieldException):
return handle_non_field_error(exc, context)
elif isinstance(exc, APIException):
return handle_api_exception(exc, context)
else:
return None
def weather(request, city):
"""
Returns weather information from a particular city
"""
if city is None:
return APIException("Missing city parameter")
try:
weather_provider_class = utils.get_setting('WEATHER_PROVIDER')
weather_provider = utils.module_member(weather_provider_class)
provider_instance = weather_provider()
res = provider_instance.get_weather_data(city)
except ApiCallFailed:
return Response(status=status.HTTP_404_NOT_FOUND)
return Response(res)
def get(self, request, *args, **kwargs):
try:
provided_token = request.META['HTTP_AUTHORIZATION']
provided_token = provided_token.replace('Token ', '')
token = Token.objects.select_related('user').get(key=provided_token)
except Exception as e:
# modify the original exception response
raise exceptions.APIException("Token not valid.")
if not token.user.is_active:
# can also modify this exception message
return Response(error_response("User inactive or deleted."))
sport = int(request.query_params.get('sport', ''))
if sport == 0:
swimmer = Swimmer.objects.get(type=request.user.id)
user_serialized = SwimmerSerializer(swimmer)
elif sport == 1:
runner = Runner.objects.get(type=request.user.id)
user_serialized = RunnerSerializer(runner)
return Response(success_response(user_serialized.data))
def _share(self):
content = self.get_object()
try:
share = content.share(self.request.user.profile)
except ValidationError as e:
raise exceptions.ValidationError(e.message)
except Exception:
raise exceptions.APIException("Unknown error when creating share.")
return Response({"status": "ok", "content_id": share.id}, status=HTTP_201_CREATED)
def _unshare(self):
content = self.get_object()
try:
content.unshare(self.request.user.profile)
except ValidationError as e:
raise exceptions.ValidationError(e.message)
except Exception:
raise exceptions.APIException("Unknown error when creating share.")
return Response({"status": "ok"}, status=HTTP_204_NO_CONTENT)
def stars_top_employee_lists(request, top_number, kind, id):
"""
Returns stars top {top_number} list according to {kind} (category, keyword) {id} (kind_id)
---
serializer: stars.serializers.StarTopEmployeeLists
responseMessages:
- code: 401
message: Unauthorized. Authentication credentials were not provided. Invalid token.
- code: 403
message: Forbidden, authentication credentials were not provided
- code: 404
message: Not found
- code: 412
message: Precondition failed, kind should be category or subcategory
"""
try:
if request.method == 'GET':
if kind == 'category':
top_list = Star.objects.filter(category__id=id).values(
'to_user__pk',
'to_user__username',
'to_user__first_name',
'to_user__last_name',
'to_user__level'
'to_user__avatar').annotate(num_stars=Count('to_user')).order_by('-num_stars')[:top_number]
elif kind == 'keyword':
top_list = Star.objects.filter(keyword__id=id).values(
'to_user__pk',
'to_user__username',
'to_user__first_name',
'to_user__last_name',
'to_user__level',
'to_user__avatar').annotate(num_stars=Count('to_user')).order_by('-num_stars')[:top_number]
else:
return Response(status=status.HTTP_412_PRECONDITION_FAILED)
serializer = StarTopEmployeeLists(top_list, many=True)
return Response(serializer.data, status=status.HTTP_200_OK)
except Exception as e:
raise APIException(e)
def get(self, request, kind, format=None):
"""
List all employees
---
serializer: administrator.serializers.EmployeeSerializer
parameters:
- name: pagination
required: false
type: string
paramType: query
- name: quantity
required: false
type: string
paramType: query
"""
employee_list = Employee.objects.filter(is_active=True, is_base_profile_complete=True).order_by('-' + kind)
if request.GET.get('quantity'):
try:
quantity = request.GET.get('quantity')
employee_list = employee_list[:quantity]
except Exception as e:
raise APIException(e)
if request.GET.get('pagination'):
pagination = request.GET.get('pagination')
if pagination == 'true':
paginator = AdministratorPagination()
results = paginator.paginate_queryset(employee_list, request)
serializer = EmployeeTopSerializer(results, many=True)
return paginator.get_paginated_response(serializer.data)
else:
return Response(status=status.HTTP_400_BAD_REQUEST)
else:
serializer = EmployeeTopSerializer(employee_list, many=True)
return Response(serializer.data, status=status.HTTP_200_OK)
def perform_create(self, serializer):
'''Add a new tileset
When adding a new dataset, we need to enforce permissions as well as
other rules like the uniqueness of uuids.
Args:
serializer (tilsets.serializer.TilesetSerializer): The serializer
to use to save the request.
'''
if 'uid' in self.request.data:
try:
self.queryset.get(uuid=self.request.data['uid'])
# this uid already exists, return an error
raise rfe.APIException("UID already exists")
except tm.Tileset.DoesNotExist:
uid = self.request.data['uid']
else:
uid = slugid.nice().decode('utf-8')
if 'filetype' not in self.request.data:
raise rfe.APIException('Missing filetype')
datafile_name = self.request.data.get('datafile').name
if 'name' in self.request.data:
name = self.request.data['name']
else:
name = op.split(datafile_name)[1]
if self.request.user.is_anonymous:
# can't create a private dataset as an anonymous user
serializer.save(
owner=gu.get_anonymous_user(),
private=False,
name=name,
uuid=uid
)
else:
serializer.save(owner=self.request.user, name=name, uuid=uid)
def server_error(request):
raise APIException('APIException')
def get_object_or_404(self, pk):
queryset = self.filter_queryset(self.get_queryset())
filter_kwargs = {self.lookup_field: pk}
try:
return get_object_or_404(queryset, **filter_kwargs)
except Http404:
# transform Http404 into an APIException
raise NotFound
def has_view_permissions(self, path, method, view):
"""
Return `True` if the incoming request has the correct view permissions.
"""
if view.request is None:
return True
try:
view.check_permissions(view.request)
except (exceptions.APIException, Http404, PermissionDenied):
return False
return True
def show_form_for_method(self, view, method, request, obj):
"""
Returns True if a form should be shown for this method.
"""
if method not in view.allowed_methods:
return # Not a valid method
try:
view.check_permissions(request)
if obj is not None:
view.check_object_permissions(request, obj)
except exceptions.APIException:
return False # Doesn't have permissions
return True
def exception_handler(exc, context):
"""
Returns the response that should be used for any given exception.
By default we handle the REST framework `APIException`, and also
Django's built-in `Http404` and `PermissionDenied` exceptions.
Any unhandled exceptions may return `None`, which will cause a 500 error
to be raised.
"""
if isinstance(exc, exceptions.APIException):
headers = {}
if getattr(exc, 'auth_header', None):
headers['WWW-Authenticate'] = exc.auth_header
if getattr(exc, 'wait', None):
headers['Retry-After'] = '%d' % exc.wait
if isinstance(exc.detail, (list, dict)):
data = exc.detail
else:
data = {'detail': exc.detail}
set_rollback()
return Response(data, status=exc.status_code, headers=headers)
elif isinstance(exc, Http404):
msg = _('Not found.')
data = {'detail': six.text_type(msg)}
set_rollback()
return Response(data, status=status.HTTP_404_NOT_FOUND)
elif isinstance(exc, PermissionDenied):
msg = _('Permission denied.')
data = {'detail': six.text_type(msg)}
set_rollback()
return Response(data, status=status.HTTP_403_FORBIDDEN)
return None
def determine_actions(self, request, view):
"""
For generic class based views we return information about
the fields that are accepted for 'PUT' and 'POST' methods.
"""
actions = {}
for method in {'PUT', 'POST'} & set(view.allowed_methods):
view.request = clone_request(request, method)
try:
# Test global permissions
if hasattr(view, 'check_permissions'):
view.check_permissions(view.request)
# Test object permissions
if method == 'PUT' and hasattr(view, 'get_object'):
view.get_object()
except (exceptions.APIException, PermissionDenied, Http404):
pass
else:
# If user has appropriate permissions for the view, include
# appropriate metadata about the fields that should be supplied.
serializer = view.get_serializer()
actions[method] = self.get_serializer_info(serializer)
finally:
view.request = request
return actions
def add_player(self, request, pk=None):
game = get_object_or_404(Game, pk=pk)
body = request.data
player = get_object_or_404(Player, pk=body["id"])
with transaction.atomic():
relation = GamePlayerRelation(game=game, player=player)
relation.save()
if GamePlayerRelation.objects.filter(game=game).count() > 6:
raise APIException(detail="Game is already full", code=400)
return redirect(reverse("frisbeer:games-detail", args=[pk]))
def create_teams(self, request, pk=None):
game = get_object_or_404(Game, pk=pk)
if GamePlayerRelation.objects.filter(game=game).count() != 6:
raise APIException("Game needs 6 players before teams can be created", code=400)
force = request.data.get("re_create", False)
game.create_teams()
game.state = Game.READY
game.save()
print("Created")
return redirect(reverse("frisbeer:games-detail", args=[pk]))
def show_form_for_method(self, view, method, request, obj):
"""
Returns True if a form should be shown for this method.
"""
if method not in view.allowed_methods:
return # Not a valid method
try:
view.check_permissions(request)
if obj is not None:
view.check_object_permissions(request, obj)
except exceptions.APIException:
return False # Doesn't have permissions
return True
def exception_handler(exc, context):
"""
Returns the response that should be used for any given exception.
By default we handle the REST framework `APIException`, and also
Django's built-in `Http404` and `PermissionDenied` exceptions.
Any unhandled exceptions may return `None`, which will cause a 500 error
to be raised.
"""
if isinstance(exc, exceptions.APIException):
headers = {}
if getattr(exc, 'auth_header', None):
headers['WWW-Authenticate'] = exc.auth_header
if getattr(exc, 'wait', None):
headers['Retry-After'] = '%d' % exc.wait
if isinstance(exc.detail, (list, dict)):
data = exc.detail
else:
data = {'detail': exc.detail}
set_rollback()
return Response(data, status=exc.status_code, headers=headers)
elif isinstance(exc, Http404):
msg = _('Not found.')
data = {'detail': six.text_type(msg)}
set_rollback()
return Response(data, status=status.HTTP_404_NOT_FOUND)
elif isinstance(exc, PermissionDenied):
msg = _('Permission denied.')
data = {'detail': six.text_type(msg)}
set_rollback()
return Response(data, status=status.HTTP_403_FORBIDDEN)
# Note: Unhandled exceptions will raise a 500 error.
return None
def determine_actions(self, request, view):
"""
For generic class based views we return information about
the fields that are accepted for 'PUT' and 'POST' methods.
"""
actions = {}
for method in {'PUT', 'POST'} & set(view.allowed_methods):
view.request = clone_request(request, method)
try:
# Test global permissions
if hasattr(view, 'check_permissions'):
view.check_permissions(view.request)
# Test object permissions
if method == 'PUT' and hasattr(view, 'get_object'):
view.get_object()
except (exceptions.APIException, PermissionDenied, Http404):
pass
else:
# If user has appropriate permissions for the view, include
# appropriate metadata about the fields that should be supplied.
serializer = view.get_serializer()
actions[method] = self.get_serializer_info(serializer)
finally:
view.request = request
return actions
def authenticate(self, request):
api_settings = APISettings.get_solo()
if not api_settings.allow:
raise exceptions.PermissionDenied('API is disabled')
if request.META.get('HTTP_X_AUTHKEY') != api_settings.auth_key:
raise exceptions.AuthenticationFailed('Invalid auth key')
try:
user = User.objects.get(username=settings.DSMRREADER_REST_FRAMEWORK_API_USER)
except User.DoesNotExist:
raise exceptions.APIException('API user not found')
return (user, None)
def exception_handler(exc, context=None):
"""Returns the response that should be used for any given exception.
By default we handle the REST framework `APIException`, and also
Django's builtin `Http404` and `PermissionDenied` exceptions.
Any unhandled exceptions may return `None`, which will cause a 500 error
to be raised.
"""
if isinstance(exc, exceptions.APIException):
headers = {}
if getattr(exc, 'auth_header', None):
headers['WWW-Authenticate'] = exc.auth_header
if getattr(exc, 'wait', None):
headers['X-Throttle-Wait-Seconds'] = '%d' % exc.wait
detail = format_exception(exc)
return Response(detail, status=exc.status_code, headers=headers)
elif isinstance(exc, Http404):
return Response({'error_type': exc.__class__.__name__,
'errors': [{'message': str(exc)}]},
status=status.HTTP_404_NOT_FOUND)
elif isinstance(exc, DjangoPermissionDenied):
return Response({'error_type': exc.__class__.__name__,
'errors': [{'message': str(exc)}]},
status=status.HTTP_403_FORBIDDEN)
# Note: Unhandled exceptions will raise a 500 error.
return None
def _apply_geo_filters(self, request, queryset):
# Strictly speaking these are not queries that should be possible with a GeoReport v2
# core implementation, but as they do not require extra data in the models, it's worth it
# to have them available "for free".
bbox = request.query_params.get('bbox')
if bbox:
bbox = parse_bbox(bbox)
(long1, lat1), (long2, lat2) = bbox
queryset = queryset.filter(lat__range=(lat1, lat2))
queryset = queryset.filter(long__range=(long1, long2))
lat = request.query_params.get('lat')
lon = request.query_params.get('long')
radius = request.query_params.get('radius')
if lat and lon and radius:
try:
lat = float(lat)
lon = float(lon)
radius = float(radius)
except ValueError:
raise APIException('lat/lon/radius must all be valid decimal numbers')
if not determine_gissiness():
raise APIException('this installation is not capable of lat/lon/radius queries')
from django.contrib.gis.db.models.functions import Distance
from django.contrib.gis.geos import Point
from django.contrib.gis.measure import D
point = Point(x=lon, y=lat, srid=4326)
queryset = queryset.annotate(distance=Distance('location', point))
queryset = queryset.filter(location__distance_lte=(point, D(m=radius)))
return queryset
def create(self, validated_data, *args, **kwargs):
city = self.Meta.model.look_up_city(latitude=validated_data['latitude'],
longitude=validated_data['longitude'] ,
)
validated_data['city'] = city
place_qs = self.Meta.model.objects.filter(city=validated_data['city'],
name=validated_data['name']
)
if place_qs:
raise APIException('city: "{}"??????"{}"???,?????'.format(
validated_data['city'], validated_data['name']))
return super(PlaceSerializer, self).create(validated_data, *args, **kwargs)
def create(self, validated_data):
validated_data = {
'email': validated_data.get('email', ''),
'password': validated_data.get('password1', ''),
'gender': validated_data.get('gender'),
'age': validated_data.get('age'),
'latitude': validated_data.get('latitude'),
'hardness': validated_data.get('hardness'),
'registration_id': validated_data.get('registration_id'),
}
try:
return CustomUser.objects.create_user(**validated_data)
except Exception as e:
raise APIException({"detail": e.args})
def perform_create(self, serializer):
serializer.save()
post = Post.objects.get(pk=self.kwargs.get('post_pk'))
try:
Alarm.objects.create(post=post, comment_author=self.request.user)
post_pk = self.kwargs.get('post_pk')
post = get_object_or_404(Post, pk=post_pk)
registration_id = post.author.registration_id
is_logined = Token.objects.filter(user=post.author)
if registration_id and is_logined:
message_body = "??? ? ? '{}'...? ??? ?????".format(post.content)
messaging(message_body, post.pk, registration_id)
except Exception as e:
raise APIException({"detail": e.args})
def exception_handler(exc, context):
if isinstance(exc, exceptions.APIException):
if isinstance(exc.detail, dict):
data = exc.detail
else:
data = {'detail': exc.detail}
set_rollback()
return data
# Note: Unhandled exceptions will raise a 500 error.
return None
def handle_api_exception(exc, context):
"""
Handle the given APIException.
:param exc: The exception that was thrown.
:param context: The context in which the exception was thrown.
:return: A Django Response object.
"""
response = exception_handler(exc, context)
response.data = {
"status_code": response.status_code,
"message": "Exception thrown",
"detail": exc.detail,
"error_fields": [],
}
return response
def top(request, kind, quantity):
"""
Returns top {quantity} list, {kind} (total_score, level, last_month_score, current_month_score, last_year_score,
current_year_score)
---
serializer: employees.serializers.EmployeeTopListSerializer
responseMessages:
- code: 401
message: Unauthorized. Authentication credentials were not provided. Invalid token.
- code: 403
message: Forbidden, authentication credentials were not provided
- code: 404
message: Not found
- code: 500
message: Internal server error, cannot resolve keyword into field.
"""
try:
employee_list_filtered = []
if request.method == 'GET':
employee_list = Employee.objects.filter(is_active=True,
is_base_profile_complete=True).order_by('-' + kind)[:quantity]
if kind == 'total_score':
for employee in employee_list:
if employee.total_score > 0:
employee_list_filtered.append(employee)
serializer = EmployeeTopTotalScoreList(employee_list_filtered, many=True)
elif kind == 'level':
for employee in employee_list:
if employee.level > 0:
employee_list_filtered.append(employee)
serializer = EmployeeTopLevelList(employee_list_filtered, many=True)
elif kind == 'current_month_score':
for employee in employee_list:
if employee.current_month_score > 0:
employee_list_filtered.append(employee)
serializer = EmployeeTopCurrentMonthList(employee_list_filtered, many=True)
elif kind == 'current_year_score':
for employee in employee_list:
if employee.current_year_score > 0:
employee_list_filtered.append(employee)
serializer = EmployeeTopCurrentYearList(employee_list_filtered, many=True)
elif kind == 'last_month_score':
for employee in employee_list:
if employee.last_month_score > 0:
employee_list_filtered.append(employee)
serializer = EmployeeTopLastMonthList(employee_list_filtered, many=True)
elif kind == 'last_year_score':
for employee in employee_list:
if employee.last_year_score > 0:
employee_list_filtered.append(employee)
serializer = EmployeeTopLastYearList(employee_list_filtered, many=True)
return Response(serializer.data, status=status.HTTP_200_OK)
except Exception as e:
raise APIException(e)