Python rest_framework.status 模块，HTTP_429_TOO_MANY_REQUESTS 实例源码

def create(self, request, *args, **kwargs):
    email = request.data.get('email', None)

    try:
      user = self.get_queryset().get(email__iexact=email)
    except:
      user = None

    if user:
      # Allow only 5 requests per hour
      limit = 5
      now = timezone.now()
      to_check = (now - relativedelta(hours=1)).replace(tzinfo=timezone.utc)
      tokens = models.PasswordRecoveryToken.objects.filter(user=user, created_date__gte=to_check, channel__slug=request.channel)

      if tokens.count() >= limit:
        will_release = tokens.order_by('-created_date')[limit-1].created_date + relativedelta(hours=1)
        seconds = abs((will_release - now).seconds)
        return response.Response({'success': False, 'message': 'Five tokens generated last hour.', 'try_again_in': seconds}, status=status.HTTP_429_TOO_MANY_REQUESTS)

      token = models.PasswordRecoveryToken.objects.create(user=user, object_channel=request.channel)

    return response.Response({'success': True, 'message': 'Token requested successfully(if user exists).'})

def test_search_with_search_anon_rate_throttle(self, mock_essearch):
        url = reverse('complaint_search:search')
        mock_essearch.return_value = 'OK'
        SearchAnonRateThrottle.rate = self.orig_search_anon_rate
        ExportUIRateThrottle.rate = self.orig_export_ui_rate
        ExportAnonRateThrottle.rate = self.orig_export_anon_rate
        limit = int(self.orig_search_anon_rate.split('/')[0])
        for _ in range(limit):
            response = self.client.get(url)
            self.assertEqual(response.status_code, status.HTTP_200_OK)
            self.assertEqual('OK', response.data)

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_429_TOO_MANY_REQUESTS)
        self.assertIsNotNone(response.data.get('detail'))
        self.assertIn("Request was throttled", response.data.get('detail'))
        self.assertEqual(limit, mock_essearch.call_count)
        self.assertEqual(20, limit)

def test_search_with_export_anon_rate_throttle(self, mock_essearch):
        url = reverse('complaint_search:search')
        mock_essearch.return_value = 'OK'
        SearchAnonRateThrottle.rate = self.orig_search_anon_rate
        ExportUIRateThrottle.rate = self.orig_export_ui_rate
        ExportAnonRateThrottle.rate = self.orig_export_anon_rate
        limit = int(self.orig_export_anon_rate.split('/')[0])
        for i in range(limit):
            response = self.client.get(url, {"format": "csv"})
            self.assertEqual(response.status_code, status.HTTP_200_OK)
            self.assertTrue(isinstance(response, StreamingHttpResponse))

        response = self.client.get(url, {"format": "csv"})
        self.assertEqual(response.status_code, status.HTTP_429_TOO_MANY_REQUESTS)
        self.assertIsNotNone(response.data.get('detail'))
        self.assertIn("Request was throttled", response.data.get('detail'))
        self.assertEqual(limit, mock_essearch.call_count)
        self.assertEqual(2, limit)

def test_search_with_export_ui_rate_throttle(self, mock_essearch):
        url = reverse('complaint_search:search')
        mock_essearch.return_value = 'OK'
        SearchAnonRateThrottle.rate = self.orig_search_anon_rate
        ExportUIRateThrottle.rate = self.orig_export_ui_rate
        ExportAnonRateThrottle.rate = self.orig_export_anon_rate
        limit = int(self.orig_export_ui_rate.split('/')[0])
        for _ in range(limit):
            response = self.client.get(url, {"format": "csv"}, HTTP_REFERER=_CCDB_UI_URL)
            self.assertEqual(response.status_code, status.HTTP_200_OK)
            self.assertTrue(isinstance(response, StreamingHttpResponse))

        response = self.client.get(url, {"format": "csv"}, HTTP_REFERER=_CCDB_UI_URL)
        self.assertEqual(response.status_code, status.HTTP_429_TOO_MANY_REQUESTS)
        self.assertIsNotNone(response.data.get('detail'))
        self.assertIn("Request was throttled", response.data.get('detail'))
        self.assertEqual(limit, mock_essearch.call_count)
        self.assertEqual(6, limit)

def test_document_with_document_anon_rate_throttle(self, mock_esdocument):
        url = reverse('complaint_search:complaint', kwargs={"id": "123456"})
        mock_esdocument.return_value = 'OK'
        DocumentAnonRateThrottle.rate = self.orig_document_anon_rate
        limit = int(self.orig_document_anon_rate.split('/')[0])
        for i in range(limit):
            response = self.client.get(url)
            self.assertEqual(response.status_code, status.HTTP_200_OK)
            self.assertEqual('OK', response.data)

        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_429_TOO_MANY_REQUESTS)
        self.assertIsNotNone(response.data.get('detail'))
        self.assertIn("Request was throttled", response.data.get('detail'))
        self.assertEqual(limit, mock_esdocument.call_count)
        self.assertEqual(5, limit)

def too_many_requests():
    STATUS["status"] = status.HTTP_429_TOO_MANY_REQUESTS
    STATUS["message"] = "Reached max limit for the day."
    return (STATUS, STATUS["status"])