Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
time.sleep(sleep_sec)
continue
# readonly
if 'code' in json_data['error'] and json_data['error']['code'] == 'readonly':
print('Wikidata currently is in readonly mode, waiting for {} seconds'.format(retry_after))
time.sleep(retry_after)
continue
# there is no error or waiting. break out of this loop and parse response
break
else:
# the first time I've ever used for - else!!
# else executes if the for loop completes normally. i.e. does not encouter a `break`
# in this case, that means it tried this api call 10 times
raise WDApiError(response.json() if response else dict())
return json_data
:return: True if write did not throw an exception, returns the exception otherwise
"""
if wd_item.require_write:
if wd_item.create_new_item:
msg = "CREATE"
else:
msg = "UPDATE"
else:
msg = "SKIP"
try:
if write:
wd_item.write(login=login, edit_summary=edit_summary)
wdi_core.WDItemEngine.log("INFO", format_msg(record_id, record_prop, wd_item.wd_item_id, msg) + ";" + str(
wd_item.lastrevid))
except wdi_core.WDApiError as e:
print(e)
wdi_core.WDItemEngine.log("ERROR",
format_msg(record_id, record_prop, wd_item.wd_item_id, json.dumps(e.wd_error_msg),
type(e)))
return e
except Exception as e:
print(e)
wdi_core.WDItemEngine.log("ERROR", format_msg(record_id, record_prop, wd_item.wd_item_id, str(e), type(e)))
return e
return True
else:
payload.update({u'id': self.wd_item_id})
try:
json_data = self.mediawiki_api_call('POST', self.mediawiki_api_url, session=login.get_session(),
max_retries=max_retries, retry_after=retry_after,
headers=headers, data=payload)
if 'error' in json_data and 'messages' in json_data['error']:
error_msg_names = set(x.get('name') for x in json_data["error"]['messages'])
if 'wikibase-validator-label-with-description-conflict' in error_msg_names:
raise NonUniqueLabelDescriptionPairError(json_data)
else:
raise WDApiError(json_data)
elif 'error' in json_data.keys():
raise WDApiError(json_data)
except Exception:
print('Error while writing to Wikidata')
raise
# after successful write, update this object with latest json, QID and parsed data types.
self.create_new_item = False
self.wd_item_id = json_data['entity']['id']
self.parse_wd_json(wd_json=json_data['entity'])
self.data = []
if "success" in json_data and "entity" in json_data and "lastrevid" in json_data["entity"]:
self.lastrevid = json_data["entity"]["lastrevid"]
time.sleep(.5)
return self.wd_item_id
class WDApiError(Exception):
def __init__(self, wd_error_message):
"""
Base class for Wikidata error handling
:param wd_error_message: The error message returned by the WD API
:type wd_error_message: A Python json representation dictionary of the error message
:return:
"""
self.wd_error_msg = wd_error_message
def __str__(self):
return repr(self.wd_error_msg)
class NonUniqueLabelDescriptionPairError(WDApiError):
def __init__(self, wd_error_message):
"""
This class handles errors returned from the WD API due to an attempt to create an item which has the same
label and description as an existing item in a certain language.
:param wd_error_message: An WD API error mesage containing 'wikibase-validator-label-with-description-conflict'
as the message name.
:type wd_error_message: A Python json representation dictionary of the error message
:return:
"""
self.wd_error_msg = wd_error_message
def get_language(self):
"""
:return: Returns a 2 letter Wikidata language string, indicating the language which triggered the error
"""
return self.wd_error_msg['error']['messages'][0]['parameters'][1]
class WDApiError(Exception):
def __init__(self, wd_error_message):
"""
Base class for Wikidata error handling
:param wd_error_message: The error message returned by the WD API
:type wd_error_message: A Python json representation dictionary of the error message
:return:
"""
self.wd_error_msg = wd_error_message
def __str__(self):
return repr(self.wd_error_msg)
class NonUniqueLabelDescriptionPairError(WDApiError):
def __init__(self, wd_error_message):
"""
This class handles errors returned from the WD API due to an attempt to create an item which has the same
label and description as an existing item in a certain language.
:param wd_error_message: An WD API error mesage containing 'wikibase-validator-label-with-description-conflict'
as the message name.
:type wd_error_message: A Python json representation dictionary of the error message
:return:
"""
self.wd_error_msg = wd_error_message
def get_language(self):
"""
:return: Returns a 2 letter Wikidata language string, indicating the language which triggered the error
"""
return self.wd_error_msg['error']['messages'][0]['parameters'][1]
:return: True if write did not throw an exception, returns the exception otherwise
"""
if wd_item.require_write:
if wd_item.create_new_item:
msg = "CREATE"
else:
msg = "UPDATE"
else:
msg = "SKIP"
try:
if write:
wd_item.write(login=login, edit_summary=edit_summary)
wdi_core.WDItemEngine.log("INFO", format_msg(record_id, record_prop, wd_item.wd_item_id, msg) + ";" + str(
wd_item.lastrevid))
except WDApiError as e:
print(e)
wdi_core.WDItemEngine.log("ERROR",
format_msg(record_id, record_prop, wd_item.wd_item_id, json.dumps(e.wd_error_msg),
type(e)))
return e
except Exception as e:
print(e)
wdi_core.WDItemEngine.log("ERROR", format_msg(record_id, record_prop, wd_item.wd_item_id, str(e), type(e)))
return e
return True
# pprint.pprint(json_data)
if 'error' in json_data.keys() and 'code' in json_data['error'] \
and json_data['error']['code'] == 'readonly':
print('Wikidata currently is in readonly mode, waiting for 60 seconds')
time.sleep(60)
return self.write(login=login)
if 'error' in json_data.keys() and 'messages' in json_data['error']:
if 'wikibase-validator-label-with-description-conflict' == json_data['error']['messages'][0]['name']:
raise NonUniqueLabelDescriptionPairError(json_data)
else:
raise WDApiError(json_data)
elif 'error' in json_data.keys():
raise WDApiError(json_data)
except Exception as e:
print('Error while writing to Wikidata')
raise e
# after successful write, update this object with latest json, QID and parsed data types.
self.create_new_item = False
self.wd_item_id = json_data['entity']['id']
self.parse_wd_json(wd_json=json_data['entity'])
self.data = []
return self.wd_item_id