How to use the quandl.connection.Connection.request function in Quandl

To help you get started, we’ve selected a few Quandl 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 quandl / quandl-python / test / test_connection.py View on Github external
def test_parse_error(self, request_method):
        ApiConfig.retry_backoff_factor = 0
        httpretty.register_uri(getattr(httpretty, request_method),
                               "https://www.quandl.com/api/v3/databases",
                               body="not json", status=500)
        self.assertRaises(
            QuandlError, lambda: Connection.request(request_method, 'databases'))
github quandl / quandl-python / test / test_retries.py View on Github external
def test_correct_response_returned_if_retries_succeed(self):
        ApiConfig.number_of_retries = 3
        ApiConfig.retry_status_codes = [self.error_response.status]

        mock_responses = [self.error_response] + [self.error_response] + [self.success_response]
        httpretty.register_uri(httpretty.GET,
                               "https://www.quandl.com/api/v3/databases",
                               responses=mock_responses)

        response = Connection.request('get', 'databases')
        self.assertEqual(response.json(), self.datatable)
        self.assertEqual(response.status_code, self.success_response.status)
github quandl / quandl-python / test / test_retries.py View on Github external
def test_correct_response_exception_raised_for_errors_not_in_retry_status_codes(self):
        ApiConfig.retry_status_codes = []
        mock_responses = [self.error_response]
        httpretty.register_uri(httpretty.GET,
                               "https://www.quandl.com/api/v3/databases",
                               responses=mock_responses)

        self.assertRaises(InternalServerError, Connection.request, 'get', 'databases')
github quandl / quandl-python / test / test_connection.py View on Github external
            QuandlError, lambda: Connection.request(request_method, 'databases'))
github quandl / quandl-python / quandl / model / datatable.py View on Github external
def _request_file_info(self, file_or_folder_path, **options):
        url = self._download_request_path()
        code_name = self.code
        options['params']['qopts.export'] = 'true'

        request_type = RequestType.get_request_type(url, **options)

        updated_options = Util.convert_options(request_type=request_type, **options)

        r = Connection.request(request_type, url, **updated_options)

        response_data = r.json()

        file_info = response_data['datatable_bulk_download']['file']

        status = file_info['status']

        if status == 'fresh':
            file_link = file_info['link']
            self._download_file_with_link(file_or_folder_path, file_link, code_name)
            return True
        else:
            return False
github quandl / quandl-python / quandl / operations / list.py View on Github external
def all(cls, **options):
        if 'params' not in options:
            options['params'] = {}
        path = Util.constructed_path(cls.list_path(), options['params'])
        r = Connection.request('get', path, **options)
        response_data = r.json()
        Util.convert_to_dates(response_data)
        resource = cls.create_list_from_response(response_data)
        return resource
github quandl / quandl-python / quandl / operations / list.py View on Github external
def page(cls, datatable, **options):
        params = {'id': str(datatable.code)}
        path = Util.constructed_path(datatable.default_path(), params)

        request_type = RequestType.get_request_type(path, **options)

        updated_options = Util.convert_options(request_type=request_type, **options)

        r = Connection.request(request_type, path, **updated_options)

        response_data = r.json()
        Util.convert_to_dates(response_data)
        resource = cls.create_datatable_list_from_response(response_data)
        return resource
github quandl / quandl-python / quandl / model / database.py View on Github external
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