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_build_request(self, request_method, mock):
ApiConfig.api_key = 'api_token'
ApiConfig.api_version = '2015-04-09'
params = {'per_page': 10, 'page': 2}
headers = {'x-custom-header': 'header value'}
Connection.request(request_method, 'databases', headers=headers, params=params)
expected = call(request_method, 'https://www.quandl.com/api/v3/databases',
headers={'x-custom-header': 'header value',
'x-api-token': 'api_token',
'accept': ('application/json, '
'application/vnd.quandl+json;version=2015-04-09'),
'request-source': 'python',
'request-source-version': VERSION},
params={'per_page': 10, 'page': 2})
self.assertEqual(mock.call_args, expected)
def setUp(self):
# ensure api key is not set
ApiConfig.api_key = None
setupDatasetsTest(self, httpretty)
def test_setting_api_key_config(self):
mock_connection = Mock(wraps=Connection)
with patch('quandl.connection.Connection.execute_request',
new=mock_connection.execute_request) as mock:
ApiConfig.api_key = 'api_key_configured'
get('NSE/OIL')
# extract the headers passed to execute_request
actual_request_headers = mock.call_args[1]['headers']
self.assertEqual(actual_request_headers['x-api-token'], 'api_key_configured')
def test_sets_api_key_using_authtoken_arg(self):
get('NSE/OIL', authtoken='api_key')
self.assertEqual(ApiConfig.api_key, 'api_key')
def setUp(self):
datatable = {'datatable': DatatableFactory.build(
vendor_code='AUSBS', datatable_code='D')}
self.datatable = Datatable(datatable['datatable']['vendor_code'] + '/' +
datatable['datatable']['datatable_code'], datatable['datatable'])
ApiConfig.api_key = 'api_token'
ApiConfig.api_version = '2015-04-09'
def setUp(self):
httpretty.enable()
httpretty.register_uri(httpretty.GET,
re.compile(
'https://www.quandl.com/api/v3/databases/*'),
adding_headers={
'Location': 'https://www.blah.com/download/db.zip'
},
body='{}', status=302)
httpretty.register_uri(httpretty.GET,
re.compile('https://www.blah.com/'), body='{}')
database = {'database': DatabaseFactory.build(database_code='NSE')}
self.database = Database(database['database']['database_code'], database['database'])
ApiConfig.api_key = 'api_token'
ApiConfig.api_version = '2015-04-09'
def request(cls, http_verb, url, **options):
if 'headers' in options:
headers = options['headers']
else:
headers = {}
accept_value = 'application/json'
if ApiConfig.api_version:
accept_value += ", application/vnd.quandl+json;version=%s" % ApiConfig.api_version
headers = Util.merge_to_dicts({'accept': accept_value,
'request-source': 'python',
'request-source-version': VERSION}, headers)
if ApiConfig.api_key:
headers = Util.merge_to_dicts({'x-api-token': ApiConfig.api_key}, headers)
options['headers'] = headers
abs_url = '%s/%s' % (ApiConfig.api_base, url)
return cls.execute_request(http_verb, abs_url, **options)
data = None
page_count = 0
while True:
next_options = copy.deepcopy(options)
next_data = Datatable(datatable_code).data(params=next_options)
if data is None:
data = next_data
else:
data.extend(next_data)
if page_count >= ApiConfig.page_limit:
raise LimitExceededError(
Message.WARN_DATA_LIMIT_EXCEEDED % (datatable_code,
ApiConfig.api_key
)
)
next_cursor_id = next_data.meta['next_cursor_id']
if next_cursor_id is None:
break
elif paginate is not True and next_cursor_id is not None:
warnings.warn(Message.WARN_PAGE_LIMIT_EXCEEDED, UserWarning)
break
page_count = page_count + 1
options['qopts.cursor_id'] = next_cursor_id
return data.to_pandas()
def read_key(filename=None):
if filename is None:
filename = os.path.join(os.path.expanduser('~'), '.quandl_apikey')
with open(filename, 'r') as f:
apikey = f.read()
if not apikey:
raise ValueError("File '{:s}' is empty.".format(filename))
ApiConfig.api_key = apikey
def bulk_download_url(self, **options):
url = self._bulk_download_path()
url = ApiConfig.api_base + '/' + url
if 'params' not in options:
options['params'] = {}
if ApiConfig.api_key:
options['params']['api_key'] = ApiConfig.api_key
if ApiConfig.api_version:
options['params']['api_version'] = ApiConfig.api_version
if list(options.keys()):
url += '?' + urlencode(options['params'])
return url