How to use the b2sdk.exception.TooManyRequests function in b2sdk

To help you get started, we’ve selected a few b2sdk examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github Backblaze / b2-sdk-python / test / test_b2http.py View on Github external
def test_too_many_requests_retry_header_combination_one(self):
        # If the first response didn't have a header, second one has, and third one doesn't have, what should happen?

        with patch('time.sleep') as mock_time:
            fcn = MagicMock()
            fcn.side_effect = [
                TooManyRequests(retry_after_seconds=2),
                TooManyRequests(),
                TooManyRequests(retry_after_seconds=2),
                self.response,
            ]
            self.assertIs(self.response, _translate_and_retry(fcn, 4))
            self.assertEqual([call(2), call(1.5), call(2)], mock_time.mock_calls)
github Backblaze / b2-sdk-python / test / test_b2http.py View on Github external
def test_too_many_requests_failed_after_sleep(self):
        with patch('time.sleep') as mock_time:
            fcn = MagicMock()
            fcn.side_effect = [
                TooManyRequests(retry_after_seconds=2),
                TooManyRequests(retry_after_seconds=5),
            ]
            with self.assertRaises(TooManyRequests):
                _translate_and_retry(fcn, 2)
            self.assertEqual([call(2)], mock_time.mock_calls)
github Backblaze / b2-sdk-python / test / test_b2http.py View on Github external
def test_too_many_requests_retry_header_combination_two(self):
        # If the first response had header, and the second did not, but the third has header again, what should happen?

        with patch('time.sleep') as mock_time:
            fcn = MagicMock()
            fcn.side_effect = [
                TooManyRequests(),
                TooManyRequests(retry_after_seconds=5),
                TooManyRequests(),
                self.response,
            ]
            self.assertIs(self.response, _translate_and_retry(fcn, 4))
            self.assertEqual([call(1.0), call(5), call(2.25)], mock_time.mock_calls)
github Backblaze / b2-sdk-python / test / test_b2http.py View on Github external
def test_too_many_requests(self):
        response = MagicMock()
        response.status_code = 429
        response.headers = {'retry-after': 1}
        response.content = six.b(
            '{"status": 429, "code": "Too Many requests", "message": "retry after some time"}'
        )
        with self.assertRaises(TooManyRequests):
            _translate_errors(lambda: response)
github Backblaze / b2-sdk-python / test / test_b2http.py View on Github external
def test_too_many_requests_retry_header_combination_two(self):
        # If the first response had header, and the second did not, but the third has header again, what should happen?

        with patch('time.sleep') as mock_time:
            fcn = MagicMock()
            fcn.side_effect = [
                TooManyRequests(),
                TooManyRequests(retry_after_seconds=5),
                TooManyRequests(),
                self.response,
            ]
            self.assertIs(self.response, _translate_and_retry(fcn, 4))
            self.assertEqual([call(1.0), call(5), call(2.25)], mock_time.mock_calls)
github Backblaze / b2-sdk-python / test / test_b2http.py View on Github external
def test_too_many_requests_works_after_sleep(self):
        with patch('time.sleep') as mock_time:
            fcn = MagicMock()
            fcn.side_effect = [TooManyRequests(retry_after_seconds=2), self.response]
            self.assertIs(self.response, _translate_and_retry(fcn, 3))
            self.assertEqual([call(2)], mock_time.mock_calls)
github Backblaze / b2-sdk-python / test / test_b2http.py View on Github external
def test_too_many_requests_retry_header_combination_one(self):
        # If the first response didn't have a header, second one has, and third one doesn't have, what should happen?

        with patch('time.sleep') as mock_time:
            fcn = MagicMock()
            fcn.side_effect = [
                TooManyRequests(retry_after_seconds=2),
                TooManyRequests(),
                TooManyRequests(retry_after_seconds=2),
                self.response,
            ]
            self.assertIs(self.response, _translate_and_retry(fcn, 4))
            self.assertEqual([call(2), call(1.5), call(2)], mock_time.mock_calls)
github Backblaze / b2-sdk-python / test / test_b2http.py View on Github external
def test_too_many_requests_failed_after_sleep(self):
        with patch('time.sleep') as mock_time:
            fcn = MagicMock()
            fcn.side_effect = [
                TooManyRequests(retry_after_seconds=2),
                TooManyRequests(retry_after_seconds=5),
            ]
            with self.assertRaises(TooManyRequests):
                _translate_and_retry(fcn, 2)
            self.assertEqual([call(2)], mock_time.mock_calls)
github Backblaze / b2-sdk-python / b2sdk / exception.py View on Github external
return UploadTokenUsedConcurrently(token)
        return BadRequest(message, code)
    elif status == 400:
        return BadRequest(message, code)
    elif status == 401 and code in ("bad_auth_token", "expired_auth_token"):
        return InvalidAuthToken(message, code)
    elif status == 401:
        return Unauthorized(message, code)
    elif status == 403 and code == "storage_cap_exceeded":
        return StorageCapExceeded()
    elif status == 409:
        return Conflict()
    elif status == 416 and code == "range_not_satisfiable":
        return UnsatisfiableRange()
    elif status == 429:
        return TooManyRequests(retry_after_seconds=response_headers.get('retry-after'))
    elif 500 <= status < 600:
        return ServiceError('%d %s %s' % (status, code, message))
    return UnknownError('%d %s %s' % (status, code, message))
github Backblaze / b2-sdk-python / b2sdk / exception.py View on Github external
def __init__(self, retry_after_seconds=None):
        super(TooManyRequests, self).__init__()
        self.retry_after_seconds = retry_after_seconds