Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
:param write: If `False`, do not actually perform write. Action will be logged as if write had occured
:type write: bool
: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
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
:param write: If `False`, do not actually perform write. Action will be logged as if write had occured
:type write: bool
: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
if search_results['success'] != 1:
raise WDSearchError('WD search failed')
elif len(search_results['search']) == 0:
return []
else:
id_list = []
id_labels = []
for i in search_results['search']:
id_list.append(i['id'])
id_labels.append(i['label'])
return id_list
except requests.HTTPError as e:
WDItemEngine.log('ERROR', str(e))
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
def create(self, login=None):
"""
Attempt to create new item. Returns `None` if creation fails.
"""
if login is None:
raise ValueError("login required to create item")
try:
self.parse_metadata()
self.make_statements()
item = wdi_core.WDItemEngine(item_name=self.meta['title'], data=self.statements,
domain="scientific_article")
except Exception as e:
msg = format_msg(self.ext_id, self.id_type, None, str(e), type(e))
print(msg)
wdi_core.WDItemEngine.log("ERROR", msg)
return None
item.set_label(self.meta['title'])
description = ', '.join(self.descriptions[x] for x in self.meta['pubtype_qid'])
item.set_description(description, lang='en')
write_success = try_write(item, self.ext_id, self.PROPS[self.id_types[self.id_type]], login)
if write_success:
self._cache[(self.ext_id, self.id_type)] = item.wd_item_id
return item.wd_item_id
else:
return None
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