Python rest_framework.exceptions 模块，Throttled() 实例源码

def create(self, validated_data):

        from biohub.core.conf import settings as biohub_settings
        from datetime import timedelta
        from django.utils.timezone import now

        brick = validated_data.pop('brick_name')
        author = validated_data['author']

        if Experience.objects.filter(
            author=author,
            brick=brick,
            pub_time__gte=now() - timedelta(seconds=biohub_settings.THROTTLE['experience'])
        ).exists():
            raise Throttled()

        content_serializer = validated_data.pop('content_input')
        content = content_serializer.save()
        experience = Experience.objects.create(
            brick=brick, content=content,
            author_name=author.username,
            **validated_data
        )
        return experience

def vote(self, user):

        from django.core.cache import cache
        from biohub.core.conf import settings as biohub_settings

        key = 'user_{}_vote'.format(user.id)
        if cache.get(key) is not None:
            raise Throttled()

        cache.set(key, 1, timeout=biohub_settings.THROTTLE['vote'])

        if self.author is not None and self.author.id == user.id:
            return False
        if not self.voted_users.filter(pk=user.id).exists():
            with transaction.atomic():
                self.votes += 1
                self.voted_users.add(user)
                self.save(update_fields=['votes'])
                voted_experience_signal.send(
                    sender=self.__class__, instance=self,
                    user_voted=user, current_votes=self.votes)
            return True
        return False

def throttled(self, request, wait):
        """
        If request is throttled, determine what kind of exception to raise.
        """
        raise exceptions.Throttled(wait)

def throttled(self, request, wait):
        """
        If request is throttled, determine what kind of exception to raise.
        """
        raise exceptions.Throttled(wait)

def allow_request(self, request, view):
        self.key = self.get_cache_key(request, view)
        if self.key is None:
            return True

        self.history = self.cache.get(self.key, [])
        self.now = self.timer()
        while self.history and self.history[-1][0] <= self.now - float(self.post_rate_time):
            if self.history[-1][1] == 'post':
                self.history.pop()
        if len(self.history) >= self.post_rate_num:
            return self.throttle_failure()

        if self.history and self.history[0][1] == 'duplicate':
            time_remaining = self.history[0][0] - (self.now - float(self.duplicate_lockout))
            if time_remaining > 0:
                raise Throttled(detail="You posted a duplicate comment recently.", wait=time_remaining)
            else:
                self.history.pop(0)

        if self.is_recent_duplicate(request):
            self.history.insert(0, [self.now, 'duplicate'])
            self.cache.set(self.key, self.history, float(self.duplicate_lockout))
            raise Throttled(detail=("that's a duplicate"), wait=(self.duplicate_lockout))

        return self.throttle_success()

def create(self, validated_data):

        from django.core.cache import cache
        from biohub.core.conf import settings as biohub_settings

        key = 'user_{}_rate'.format(self.context['user'].id)
        if cache.get(key) is not None:
            raise Throttled()
        cache.set(key, 1, timeout=biohub_settings.THROTTLE['rate'])

        return self.context['brick'].rate(
            self.context['user'], validated_data['score']
        )

def create(self, request, *args, **kwargs):
        from biohub.core.conf import settings as biohub_settings
        from datetime import timedelta
        from django.utils.timezone import now

        user = self.request.user
        if Post.objects.filter(
            author=user,
            pub_time__gte=now() - timedelta(seconds=biohub_settings.THROTTLE['post'])
        ).exists():
            raise exceptions.Throttled()

        return super(PostViewSet, self).create(request, *args, **kwargs)

def _check_throttle(self):
        from django.core.cache import cache
        from biohub.core.conf import settings as biohub_settings

        request = self.context['request']
        ip = get_ip_from_request(request)
        key = '{}_{}_register'.format(ip, request.session.session_key)

        if cache.get(key) is not None:
            raise Throttled()

        cache.set(key, 1, timeout=biohub_settings.THROTTLE['register'])