How to use the wikidataintegrator.wdi_core.WDApiError function in wikidataintegrator

To help you get started, we’ve selected a few wikidataintegrator examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github SuLab / WikidataIntegrator / wikidataintegrator / wdi_core.py View on Github external
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
github SuLab / WikidataIntegrator / wikidataintegrator / wdi_helpers / __init__.py View on Github external
: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
github SuLab / WikidataIntegrator / wikidataintegrator / wdi_core.py View on Github external
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
github SuLab / WikidataIntegrator / wikidataintegrator / wdi_core.py View on Github external
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]
github SuLab / WikidataIntegrator / wikidataintegrator / wdi_core.py View on Github external
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]
github SuLab / WikidataIntegrator / wikidataintegrator / wdi_helpers.py View on Github external
: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
github SuLab / WikidataIntegrator / wikidataintegrator / wdi_core.py View on Github external
# 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