Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_quandl_exceptions_no_retries(self, request_method):
ApiConfig.use_retries = False
quandl_errors = [('QELx04', 429, LimitExceededError),
('QEMx01', 500, InternalServerError),
('QEAx01', 400, AuthenticationError),
('QEPx02', 403, ForbiddenError),
('QESx03', 422, InvalidRequestError),
('QECx05', 404, NotFoundError),
('QEXx01', 503, ServiceUnavailableError),
('QEZx02', 400, QuandlError)]
httpretty.register_uri(getattr(httpretty, request_method),
"https://www.quandl.com/api/v3/databases",
responses=[httpretty.Response(body=json.dumps(
{'quandl_error':
{'code': x[0], 'message': 'something went wrong'}}),
status=x[1]) for x in quandl_errors]
)
for expected_error in quandl_errors:
self.assertRaises(
expected_error[2], lambda: Connection.request(request_method, 'databases'))
def test_bulk_download_raises_exception_when_no_path(self):
self.assertRaises(
QuandlError, lambda: self.database.bulk_download_to_file(None))
def test_bulk_download_raises_exception_when_no_path(self, request_method):
if request_method == 'POST':
RequestType.USE_GET_REQUEST = False
self.assertRaises(
QuandlError, lambda: self.datatable.download_file(None, params={}))
def handle_api_error(cls, resp):
error_body = cls.parse(resp)
# if our app does not form a proper quandl_error response
# throw generic error
if 'quandl_error' not in error_body:
raise QuandlError(http_status=resp.status_code, http_body=resp.text)
code = error_body['quandl_error']['code']
message = error_body['quandl_error']['message']
prog = re.compile('^QE([a-zA-Z])x')
if prog.match(code):
code_letter = prog.match(code).group(1)
d_klass = {
'L': LimitExceededError,
'M': InternalServerError,
'A': AuthenticationError,
'P': ForbiddenError,
'S': InvalidRequestError,
'C': NotFoundError,
'X': ServiceUnavailableError
}
}
class AuthenticationError(QuandlError):
pass
class InvalidRequestError(QuandlError):
pass
class LimitExceededError(QuandlError):
pass
class NotFoundError(QuandlError):
pass
class ServiceUnavailableError(QuandlError):
pass
class InternalServerError(QuandlError):
pass
class ForbiddenError(QuandlError):
pass
class InvalidDataError(QuandlError):
def __str__(self):
if self.http_status is None:
status_string = ''
else:
status_string = "(Status %(http_status)s) " % {"http_status": self.http_status}
if self.quandl_error_code is None:
quandl_error_string = ''
else:
quandl_error_string = "(Quandl Error %(quandl_error_code)s) " % {
"quandl_error_code": self.quandl_error_code}
return "%(ss)s%(qes)s%(qm)s" % {
"ss": status_string, "qes": quandl_error_string, "qm": self.quandl_message
}
class AuthenticationError(QuandlError):
pass
class InvalidRequestError(QuandlError):
pass
class LimitExceededError(QuandlError):
pass
class NotFoundError(QuandlError):
pass
class ServiceUnavailableError(QuandlError):
def bulk_download_to_file(self, file_or_folder_path, **options):
if not isinstance(file_or_folder_path, str):
raise QuandlError(Message.ERROR_FOLDER_ISSUE)
path_url = self._bulk_download_path()
options['stream'] = True
r = Connection.request('get', path_url, **options)
file_path = file_or_folder_path
if os.path.isdir(file_or_folder_path):
file_path = file_or_folder_path + '/' + os.path.basename(urlparse(r.url).path)
with open(file_path, 'wb') as fd:
for chunk in r.iter_content(self.BULK_CHUNK_SIZE):
fd.write(chunk)
return file_path
pass
class InternalServerError(QuandlError):
pass
class ForbiddenError(QuandlError):
pass
class InvalidDataError(QuandlError):
pass
class ColumnNotFound(QuandlError):
pass
def parse(cls, response):
try:
return response.json()
except ValueError:
raise QuandlError(http_status=response.status_code, http_body=response.text)