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
python类get_object_or_404()的实例源码
def get_object(self, prefetch=True):
"""
Returns the object the view is displaying.
You may want to override this if you need to provide non-standard
queryset lookups. Eg if objects are referenced using multiple
keyword arguments in the url conf.
"""
queryset = self.filter_queryset(self.get_queryset(prefetch=prefetch))
# Perform the lookup filtering.
lookup_url_kwarg = self.lookup_url_kwarg or self.lookup_field
assert lookup_url_kwarg in self.kwargs, (
'Expected view %s to be called with a URL keyword argument '
'named "%s". Fix your URL conf, or set the `.lookup_field` '
'attribute on the view correctly.' %
(self.__class__.__name__, lookup_url_kwarg)
)
filter_kwargs = {self.lookup_field: self.kwargs[lookup_url_kwarg]}
obj = get_object_or_404(queryset, **filter_kwargs)
# May raise a permission denied
self.check_object_permissions(self.request, obj)
return obj
def retrieve(self, request, pk):
queryset = self.get_queryset()
instance = get_object_or_404(queryset, pk=pk)
children = queryset.filter(parent=instance)
parent_serializer = self.get_serializer(instance)
parent_data = parent_serializer.data
child_serializer = self.get_serializer(children, many=True)
parent_data['children'] = child_serializer.data
return Response(parent_data)
def delete(self, request, *args, **kwargs):
try:
uuid = request.data['uuid']
except KeyError:
raise Http404
filter_params = {
'uuid': uuid,
'user': request.user
}
get_object_or_404(self.model, **filter_params).delete()
return Response(status=204)
def get(self, request, supplement_uuid):
supplement = get_object_or_404(Supplement, uuid=supplement_uuid, user=request.user)
user = request.user
serializer = SupplementLogRequestParametersSerializer(data=request.query_params)
serializer.is_valid(raise_exception=True)
params = serializer.validated_data
start_date = params['start_date']
end_date = get_current_userdate(user)
supplement_events = SupplementLog.objects.filter(user=user, supplement=supplement, time__date__gte=start_date)
builder = SupplementEventsDataframeBuilder(supplement_events)
if params['frequency'] == 'daily':
# most of the time the dataframe contains a lot of supplements, here we are only picking one
try:
series = builder.get_flat_daily_dataframe()[supplement.name]
except KeyError:
# key error for no data if the supplement was never taken during this time
series = pd.Series()
if params['complete_date_range_in_daily_frequency']:
series = force_start_end_date_to_series(user, series, start_date, end_date)
else:
df = builder.build_dataframe()
series = df['Quantity']
json_data = series.to_json(date_format='iso')
data = json.loads(json_data)
return Response(data)
def get_object(self):
formId = self.request.GET.get('formId', '')
id_string = _extract_id_string(formId)
uuid = _extract_uuid(formId)
username = self.kwargs.get('username')
obj = get_object_or_404(Instance,
xform__user__username__iexact=username,
xform__id_string__exact=id_string,
uuid=uuid)
self.check_object_permissions(self.request, obj.xform)
return obj
def create(self, request, *args, **kwargs):
if request.method.upper() == 'HEAD':
return Response(status=status.HTTP_204_NO_CONTENT,
headers=self.get_openrosa_headers(request),
template_name=self.template_name)
xform_def = request.FILES.get('form_def_file', None)
response_status = status.HTTP_201_CREATED
username = kwargs.get('username')
form_user = (username and get_object_or_404(User, username=username)) \
or request.user
if not request.user.has_perm(
'can_add_xform',
UserProfile.objects.get_or_create(user=form_user)[0]
):
raise exceptions.PermissionDenied(
detail=_(u"User %(user)s has no permission to add xforms to "
"account %(account)s" %
{'user': request.user.username,
'account': form_user.username}))
data = {}
if isinstance(xform_def, File):
do_form_upload = DoXmlFormUpload(xform_def, form_user)
dd = publish_form(do_form_upload.publish)
if isinstance(dd, XForm):
data['message'] = _(
u"%s successfully published." % dd.id_string)
else:
data['message'] = dd['text']
response_status = status.HTTP_400_BAD_REQUEST
else:
data['message'] = _(u"Missing xml file.")
response_status = status.HTTP_400_BAD_REQUEST
return Response(data, status=response_status,
headers=self.get_openrosa_headers(request,
location=False),
template_name=self.template_name)
def retrieve(self, request, pk=None):
queryset = UserProfile.objects.all()
if pk is not None:
profile = get_object_or_404(queryset, pk=pk)
else:
profile = UserProfile.objects.get(user=request.user)
serializer = UserSerializerProfile(profile)
return Response(serializer.data)
def validate(self, attrs):
email_or_username = attrs.get('email_or_username')
password = attrs.get('password')
if email_or_username and password:
# Check if user sent email
if validateEmail(email_or_username):
user_request = get_object_or_404(
User,
email=email_or_username,
)
email_or_username = user_request.username
user = authenticate(username=email_or_username, password=password)
if user:
if not user.is_active:
msg = _('User account is disabled.')
raise ValidationError(msg)
else:
msg = _('Unable to log in with provided credentials.')
raise ValidationError(msg)
else:
msg = _('Must include "email or username" and "password"')
raise ValidationError(msg)
attrs['user'] = user
return attrs
# class ProfileSerializer(serializers.ModelSerializer):
#
# class Meta:
# model = UserProfile
# exclude = ('user', 'id', 'organization')
#
def get_object(self):
"""
Incase the lookup is on an object that has been hyperlinked
then update the queryset filter appropriately
"""
if self.kwargs.get(self.lookup_field, None) is None:
raise ParseError(
'Expected URL keyword argument `%s`.' % self.lookup_field
)
queryset = self.filter_queryset(self.get_queryset())
filter_kwargs = {}
serializer = self.get_serializer()
lookup_field = self.lookup_field
if self.lookup_field in serializer.get_fields():
k = serializer.get_fields()[self.lookup_field]
if isinstance(k, serializers.HyperlinkedRelatedField):
lookup_field = '%s__%s' % (self.lookup_field, k.lookup_field)
filter_kwargs[lookup_field] = self.kwargs[self.lookup_field]
obj = get_object_or_404(queryset, **filter_kwargs)
# May raise a permission denied
self.check_object_permissions(self.request, obj)
return obj
def pre_save(self, obj):
"""
Set any attributes on the object that are implicit in the request.
"""
# pk and/or slug attributes are implicit in the URL.
lookup = self.kwargs.get(self.lookup_field, None)
pk = self.kwargs.get(self.pk_url_kwarg, None)
slug = self.kwargs.get(self.slug_url_kwarg, None)
slug_field = slug and self.slug_field or None
if lookup:
serializer = self.get_serializer()
k = serializer.get_fields()[self.lookup_field]
queryset = self.get_queryset()
queryset = self.filter_queryset(queryset)
if isinstance(k, serializers.HyperlinkedRelatedField):
filter = {}
lookup_field = '%s__%s' % (self.lookup_field, k.lookup_field)
filter[lookup_field] = lookup
k_obj = get_object_or_404(queryset, **filter)
lookup = getattr(k_obj, self.lookup_field)
setattr(obj, self.lookup_field, lookup)
if pk:
setattr(obj, 'pk', pk)
if slug:
setattr(obj, slug_field, slug)
# Ensure we clean the attributes so that we don't eg return integer
# pk using a string representation, as provided by the url conf kwarg.
if hasattr(obj, 'full_clean'):
exclude = mixins._get_validation_exclusions(
obj, pk, slug_field, self.lookup_field)
obj.full_clean(exclude)
def get_parent_object(self, pk):
parent_obj = get_object_or_404(self.parent_model, pk=pk)
self.check_object_permissions(self.request, parent_obj)
return parent_obj
def dispatch(self, request, *args, **kwargs):
self.msg = get_object_or_404(
fb_emails_models.IncomingMessage,
uuid=kwargs['msg_uuid'],
)
self.repo = self.msg.issue.repo
return super().dispatch(request, *args, **kwargs)
def get_object(self):
return get_object_or_404(
Chapter,
book__slug=self.kwargs['book'],
slug=self.kwargs['chapter'])
def get_object(self):
return get_object_or_404(
Page,
chapter__slug=self.kwargs['chapter'],
slug=self.kwargs['slug'])
def get_object(self):
pk = self.request.data.get('id')
queryset = self.filter_queryset(self.get_queryset())
obj = get_object_or_404(queryset, pk=pk)
# May raise a permission denied
self.check_object_permissions(self.request, obj)
return obj
def perform_create(self, serializer):
post = generics.get_object_or_404(Post, pk=self.kwargs['post'])
serializer.save(user=self.request.user, post=post)
def _get_object(request, nickname, pk):
# TODO DUP CODE. Must refactor this.
try:
mail = get_object_or_404(Mail, pk=pk)
can_read = mail.can_read(request)
if mail.recipient != nickname:
return None
if can_read == (True, None):
mail.read()
return mail
elif can_read == (False, {CannotReadReasons.secret_code}):
return None
return None
except Mail.DoesNotExist:
return None
def get_object(self):
"""
Returns the object the view is displaying.
You may want to override this if you need to provide non-standard
queryset lookups. Eg if objects are referenced using multiple
keyword arguments in the url conf.
"""
queryset = self.filter_queryset(self.get_queryset())
# Perform the lookup filtering.
lookup_url_kwarg = self.lookup_url_kwarg or self.lookup_field
assert lookup_url_kwarg in self.kwargs, (
'Expected view %s to be called with a URL keyword argument '
'named "%s". Fix your URL conf, or set the `.lookup_field` '
'attribute on the view correctly.' %
(self.__class__.__name__, lookup_url_kwarg)
)
try:
obj = queryset.get(**{self.lookup_field: self.kwargs[lookup_url_kwarg]})
except (ObjectDoesNotExist, ValidationError, TypeError, ValueError):
obj = get_object_or_404(queryset, nb_id=self.kwargs[lookup_url_kwarg])
# May raise a permission denied
self.check_object_permissions(self.request, obj)
return obj
def get_object(self, queryset=None):
"""Lookup user profile by pk or username"""
if self.kwargs.get(self.lookup_field, None) is None:
raise ParseError(
'Expected URL keyword argument `%s`.' % self.lookup_field
)
if queryset is None:
queryset = self.filter_queryset(self.get_queryset())
serializer = self.get_serializer()
lookup_field = self.lookup_field
if self.lookup_field in serializer.get_fields():
k = serializer.get_fields()[self.lookup_field]
if isinstance(k, serializers.HyperlinkedRelatedField):
lookup_field = '%s__%s' % (self.lookup_field, k.lookup_field)
lookup = self.kwargs[self.lookup_field]
filter_kwargs = {lookup_field: lookup}
try:
pk = int(lookup)
except (TypeError, ValueError):
pass
else:
filter_kwargs = {'user__pk': pk}
obj = get_object_or_404(queryset, **filter_kwargs)
# May raise a permission denied
self.check_object_permissions(self.request, obj)
return obj