Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
:type auth: :class:`carto.auth.APIKeyAuthClient`
:return:
:raise: CartoException
"""
try:
endpoint = (self.Meta.collection_endpoint
+ "{template_id}"). \
format(template_id=self.template_id)
if (auth is not None):
endpoint = (endpoint + "?auth_token={auth_token}"). \
format(auth_token=auth)
self.send(endpoint, "POST", json=params)
except CartoRateLimitException as e:
raise e
except Exception as e:
raise CartoException(e)
def __init__(self, response):
"""
Init method
:param response: The response rate limited by CARTO APIs
:type response: requests.models.Response class
:return:
"""
super(CartoRateLimitException, self).__init__(response.text)
self.limit = int(response.headers['Carto-Rate-Limit-Limit'])
self.remaining = int(response.headers['Carto-Rate-Limit-Remaining'])
self.retry_after = int(response.headers['Retry-After'])
self.reset = int(response.headers['Carto-Rate-Limit-Reset'])
def _get_table_columns(self):
query = '''
SELECT column_name, data_type
FROM information_schema.columns
WHERE table_name = '{table}' AND table_schema = '{schema}'
'''.format(table=self._table_name, schema=self._schema)
try:
table_info = self._context.execute_query(query)
return [Column(c['column_name'], pgtype=c['data_type']) for c in table_info['rows']]
except CartoRateLimitException as err:
raise err
except CartoException as e:
# this may happen when using the default_public API key
if str(e) == 'Access denied':
return self._get_query_columns()
else:
raise e
def download(self, query, retry_times=DEFAULT_RETRY_TIMES):
try:
return self.copy_client.copyto_stream(query.strip())
except CartoRateLimitException as err:
if retry_times > 0:
retry_times -= 1
warn('Read call rate limited. Waiting {s} seconds'.format(s=err.retry_after))
time.sleep(err.retry_after)
warn('Retrying...')
return self.download(query.strip(), retry_times=retry_times)
else:
warn(('Read call was rate-limited. '
'This usually happens when there are multiple queries being read at the same time.'))
raise err
def _copy_to(self, query, columns, retry_times):
copy_query = 'COPY ({0}) TO stdout WITH (FORMAT csv, HEADER true, NULL \'{1}\')'.format(query, PG_NULL)
try:
raw_result = self.copy_client.copyto_stream(copy_query)
except CartoRateLimitException as err:
if retry_times > 0:
retry_times -= 1
warn('Read call rate limited. Waiting {s} seconds'.format(s=err.retry_after))
time.sleep(err.retry_after)
warn('Retrying...')
return self._copy_to(query, columns, retry_times)
else:
warn(('Read call was rate-limited. '
'This usually happens when there are multiple queries being read at the same time.'))
raise err
converters = obtain_converters(columns)
parse_dates = date_columns_names(columns)
df = pd.read_csv(
raw_result,
def download(self, query, retry_times=DEFAULT_RETRY_TIMES):
try:
return self.copy_client.copyto_stream(query.strip())
except CartoRateLimitException as err:
if retry_times > 0:
retry_times -= 1
warn('Read call rate limited. Waiting {s} seconds'.format(s=err.retry_after))
time.sleep(err.retry_after)
warn('Retrying...')
return self.download(query.strip(), retry_times=retry_times)
else:
warn(('Read call was rate-limited. '
'This usually happens when there are multiple queries being read at the same time.'))
raise err
:return: response object
:rtype: Response
:raise CartoException:
"""
url = self.api_url + '/copyto'
params = {'api_key': self.api_key, 'q': query}
http_method = 'GET' if len(query) < MAX_GET_QUERY_LEN else 'POST'
try:
response = self.client.send(url,
http_method=http_method,
params=params,
stream=True)
response.raise_for_status()
except CartoRateLimitException as e:
raise e
except HTTPError as e:
if 400 <= response.status_code < 500:
# Client error, provide better reason
reason = response.json()['error'][0]
error_msg = u'%s Client Error: %s' % (response.status_code,
reason)
raise CartoException(error_msg)
else:
raise CartoException(e)
except Exception as e:
raise CartoException(e)
return response
def is_public(self):
"""Checks to see if table or table used by query has public privacy"""
public_context = get_context_with_public_creds(self._credentials)
try:
public_context.execute_query('EXPLAIN {}'.format(self.get_query()), do_post=False)
return True
except CartoRateLimitException as err:
raise err
except CartoException:
return False