Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
None
"""
id_regex = "[a-zA-Z0-9-_]{44}"
url_path = "docs.google.com/spreadsheet"
if match(id_regex, spread):
open_func = self.client.open_by_key
elif url_path in spread:
open_func = self.client.open_by_url
else:
open_func = self.client.open
try:
self.spread = open_func(spread)
self.refresh_spread_metadata()
except (SpreadsheetNotFound, NoValidUrlKeyFound, APIError) as error:
if create:
try:
self.spread = self.client.create(spread)
self.refresh_spread_metadata()
except Exception as e:
msg = "Couldn't create spreadsheet.\n" + str(e)
new_error = GspreadPandasException(msg)
elif isinstance(error, SpreadsheetNotFound) or "NOT_FOUND" in str(error):
new_error = SpreadsheetNotFound("Spreadsheet not found")
else:
new_error = error
# Raise new exception outside of except block for a python2/3 way to avoid
# "During handling of the above exception, another exception occurred"
if "new_error" in locals() and isinstance(new_error, Exception):
raise new_error
def __init__(self, response):
super(APIError, self).__init__(self._extract_text(response))
self.response = response
def get_spreadsheet(self):
try:
return self.gspread_client.open_by_key(self.doc_id).worksheet(self.tab_id)
except gspread.exceptions.SpreadsheetNotFound as e:
raise Exception("Trying to open non-existent or inaccessible spreadsheet document.")
except gspread.exceptions.WorksheetNotFound as e:
raise Exception("Trying to open non-existent sheet. Verify that the sheet name exists (%s)." % self.tab_id)
except gspread.exceptions.APIError as e:
if hasattr(e, 'response'):
error_json = e.response.json()
print(error_json)
error_status = error_json.get("error", {}).get("status")
email = self.credentials.get("client_email", "(email missing)")
if error_status == 'PERMISSION_DENIED':
raise Exception("The Service Account does not have permission to read or write on the spreadsheet document. Have you shared the spreadsheet with %s?" % email)
if error_status == 'NOT_FOUND':
raise Exception("Trying to open non-existent spreadsheet document. Verify the document id exists (%s)." % self.doc_id)
raise Exception("The Google API returned an error: %s" % e)
rows = value_matrix[1:]
except IndexError:
continue
self.inc_table_count()
yield TableData(
self.make_table_name(),
headers,
rows,
dp_extractor=self.dp_extractor,
type_hints=self._extract_type_hints(headers),
)
except gspread.exceptions.SpreadsheetNotFound:
raise OpenError("spreadsheet '{}' not found".format(self.title))
except gspread.exceptions.APIError as e:
raise APIError(e)
files=None,
headers=None):
response = getattr(self.session, method)(
endpoint,
json=json,
params=params,
data=data,
files=files,
headers=headers
)
if response.ok:
return response
else:
raise APIError(response)
key, worksheet_num = parse_query(query)
try:
spreadsheet_service = self._get_spreadsheet_service()
if is_url_key(key):
spreadsheet = spreadsheet_service.open_by_url(key)
else:
spreadsheet = spreadsheet_service.open_by_key(key)
data = parse_spreadsheet(spreadsheet, worksheet_num)
return json_dumps(data), None
except gspread.SpreadsheetNotFound:
return None, "Spreadsheet ({}) not found. Make sure you used correct id.".format(key)
except APIError as e:
return None, parse_api_error(e)
def request(*args, **kwargs):
try:
return ClientV4.request(client, *args, **kwargs)
except APIError as e:
error = str(e)
# Only retry on 100s quota breaches
if "RESOURCE_EXHAUSTED" in error and "100s" in error:
sleep(retry_delay)
return request(*args, **kwargs)
else:
error = e
if "error" in locals():
raise error
def _drive_request(
self, method="get", file_id=None, params=None, data=None, headers=None
):
url = "https://www.googleapis.com/drive/v3/files"
if file_id:
url += "/{}".format(file_id)
try:
res = self.request(method, url, params=params, json=data)
if res.text:
return res.json()
except APIError as e:
if "scopes" in e.response.text:
print(
"Your credentials don't have Drive API access, ignoring "
"drive specific functionality (Note this includes searching "
"spreadsheets by name)"
)
return {}
else:
reraise(e)