diff --git a/wikibaseintegrator/entities/baseentity.py b/wikibaseintegrator/entities/baseentity.py index c6b70724..4c0562ee 100644 --- a/wikibaseintegrator/entities/baseentity.py +++ b/wikibaseintegrator/entities/baseentity.py @@ -4,14 +4,12 @@ from copy import copy from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union -import ujson - from wikibaseintegrator import wbi_fastrun from wikibaseintegrator.datatypes import BaseDataType from wikibaseintegrator.models.claims import Claim, Claims from wikibaseintegrator.wbi_enums import ActionIfExists from wikibaseintegrator.wbi_exceptions import MissingEntityException -from wikibaseintegrator.wbi_helpers import delete_page, mediawiki_api_call_helper +from wikibaseintegrator.wbi_helpers import delete_page, edit_entity, mediawiki_api_call_helper from wikibaseintegrator.wbi_login import _Login if TYPE_CHECKING: @@ -23,8 +21,8 @@ class BaseEntity: ETYPE = 'base-entity' - def __init__(self, api: Optional['WikibaseIntegrator'] = None, title: Optional[str] = None, pageid: Optional[int] = None, lastrevid: Optional[int] = None, type: Optional[str] = None, id: Optional[str] = None, claims: Optional[Claims] = None, - is_bot: Optional[bool] = None, login: Optional[_Login] = None): + def __init__(self, api: Optional['WikibaseIntegrator'] = None, title: Optional[str] = None, pageid: Optional[int] = None, lastrevid: Optional[int] = None, + type: Optional[str] = None, id: Optional[str] = None, claims: Optional[Claims] = None, is_bot: Optional[bool] = None, login: Optional[_Login] = None): if not api: from wikibaseintegrator import WikibaseIntegrator self.api = WikibaseIntegrator() @@ -159,7 +157,8 @@ def from_json(self, json_data: Dict[str, Any]) -> BaseEntity: return self # noinspection PyMethodMayBeStatic - def _get(self, entity_id: str, login: Optional[_Login] = None, allow_anonymous: bool = True, is_bot: Optional[bool] = None, **kwargs: Any) -> Dict: # pylint: disable=no-self-use + def _get(self, entity_id: str, login: Optional[_Login] = None, allow_anonymous: bool = True, is_bot: Optional[bool] = None, + **kwargs: Any) -> Dict: # pylint: disable=no-self-use """ Retrieve an entity in json representation from the Wikibase instance @@ -192,7 +191,8 @@ def clear(self, **kwargs: Any) -> Dict[str, Any]: """ return self._write(data={}, clear=True, **kwargs) - def _write(self, data: Optional[Dict] = None, summary: Optional[str] = None, login: Optional[_Login] = None, allow_anonymous: bool = False, clear: bool = False, as_new: bool = False, is_bot: Optional[bool] = None, **kwargs: Any) -> Dict[str, Any]: + def _write(self, data: Optional[Dict] = None, summary: Optional[str] = None, login: Optional[_Login] = None, allow_anonymous: bool = False, clear: bool = False, + as_new: bool = False, is_bot: Optional[bool] = None, **kwargs: Any) -> Dict[str, Any]: """ Writes the entity JSON to the Wikibase instance and after successful write, returns the "entity" part of the response. @@ -224,42 +224,12 @@ def _write(self, data: Optional[Dict] = None, summary: Optional[str] = None, log # new_json_repr['claims'].pop(claim) # data = json.JSONEncoder().encode(new_json_repr) - if as_new: - data['id'] = None - - payload: Dict[str, Any] = { - 'action': 'wbeditentity', - 'data': ujson.dumps(data), - 'format': 'json', - 'summary': summary - } - - if not summary: - payload.pop('summary') - is_bot = is_bot if is_bot is not None else self.api.is_bot - if is_bot: - payload.update({'bot': ''}) - - if not as_new: - if clear: - payload.update({'clear': True}) - - if self.id: - payload.update({'id': self.id}) - else: - payload.update({'new': self.type}) - - if self.lastrevid: - payload.update({'baserevid': self.lastrevid}) - else: - payload.update({'new': self.type}) - self.id = None - login = login or self.api.login try: - json_result: dict = mediawiki_api_call_helper(data=payload, login=login, allow_anonymous=allow_anonymous, is_bot=is_bot, **kwargs) + json_result: dict = edit_entity(data=data, id=self.id, type=self.type, as_new=as_new, summary=summary, clear=clear, is_bot=is_bot, allow_anonymous=allow_anonymous, + login=login, **kwargs) except Exception: log.exception('Error while writing to the Wikibase instance') raise @@ -293,7 +263,8 @@ def delete(self, login: Optional[_Login] = None, allow_anonymous: bool = False, return delete_page(title=None, pageid=self.pageid, login=login, allow_anonymous=allow_anonymous, is_bot=is_bot, **kwargs) - def write_required(self, base_filter: Optional[List[BaseDataType | List[BaseDataType]]] = None, action_if_exists: ActionIfExists = ActionIfExists.REPLACE_ALL, **kwargs: Any) -> bool: + def write_required(self, base_filter: Optional[List[BaseDataType | List[BaseDataType]]] = None, action_if_exists: ActionIfExists = ActionIfExists.REPLACE_ALL, + **kwargs: Any) -> bool: fastrun_container = wbi_fastrun.get_fastrun_container(base_filter=base_filter, **kwargs) if base_filter is None: diff --git a/wikibaseintegrator/wbi_helpers.py b/wikibaseintegrator/wbi_helpers.py index 9108c044..eb60dd73 100644 --- a/wikibaseintegrator/wbi_helpers.py +++ b/wikibaseintegrator/wbi_helpers.py @@ -10,6 +10,7 @@ from urllib.parse import urlparse import requests +import ujson from requests import Session from wikibaseintegrator.wbi_backoff import wbi_backoff @@ -44,7 +45,8 @@ class BColors: default_session = requests.Session() -def mediawiki_api_call(method: str, mediawiki_api_url: Optional[str] = None, session: Optional[Session] = None, max_retries: int = 100, retry_after: int = 60, **kwargs: Any) -> Dict: +def mediawiki_api_call(method: str, mediawiki_api_url: Optional[str] = None, session: Optional[Session] = None, max_retries: int = 100, retry_after: int = 60, + **kwargs: Any) -> Dict: """ A function to call the MediaWiki API. @@ -135,7 +137,8 @@ def mediawiki_api_call(method: str, mediawiki_api_url: Optional[str] = None, ses return json_data -def mediawiki_api_call_helper(data: Dict[str, Any], login: Optional[_Login] = None, mediawiki_api_url: Optional[str] = None, user_agent: Optional[str] = None, allow_anonymous: bool = False, +def mediawiki_api_call_helper(data: Dict[str, Any], login: Optional[_Login] = None, mediawiki_api_url: Optional[str] = None, user_agent: Optional[str] = None, + allow_anonymous: bool = False, max_retries: int = 1000, retry_after: int = 60, maxlag: int = 5, is_bot: bool = False, **kwargs: Any) -> Dict: """ A simplified function to call the MediaWiki API. @@ -212,7 +215,8 @@ def mediawiki_api_call_helper(data: Dict[str, Any], login: Optional[_Login] = No @wbi_backoff() -def execute_sparql_query(query: str, prefix: Optional[str] = None, endpoint: Optional[str] = None, user_agent: Optional[str] = None, max_retries: int = 1000, retry_after: int = 60) -> Dict[str, Dict]: +def execute_sparql_query(query: str, prefix: Optional[str] = None, endpoint: Optional[str] = None, user_agent: Optional[str] = None, max_retries: int = 1000, + retry_after: int = 60) -> Dict[str, Dict]: """ Static method which can be used to execute any SPARQL query @@ -274,6 +278,64 @@ def execute_sparql_query(query: str, prefix: Optional[str] = None, endpoint: Opt raise Exception(f"No result after {max_retries} retries.") +def edit_entity(data: Dict, id: Optional[str] = None, as_new: bool = False, type: Optional[str] = None, baserevid: Optional[int] = None, summary: Optional[str] = None, + clear: bool = False, is_bot: bool = False, + login: Optional[_Login] = None, tags: Optional[List[str]] = None, site: Optional[str] = None, title: Optional[str] = None, **kwargs: Any) -> Dict: + """ + Creates a single new Wikibase entity and modifies it with serialised information. + + :param data: The serialized object that is used as the data source. A newly created entity will be assigned an 'id'. + :param id: The identifier for the entity, including the prefix. Use either id or site and title together. + :param as_new: If set, a new entity will be created. It is not allowed to have this set when id is also set. + :param type: Set this to the type of the entity to be created. One of the following values: form, item, lexeme, property, sense + :param baserevid: The numeric identifier for the revision to base the modification on. This is used for detecting conflicts during save. + :param summary: Summary for the edit. Will be prepended by an automatically generated comment. + :param clear: If set, the complete entity is emptied before proceeding. The entity will not be saved before it is filled with the "data", possibly with parts excluded. + :param is_bot: Mark this edit as bot. + :param login: A login instance + :param tags: Change tags to apply to the revision. + :param site: An identifier for the site on which the page resides. Use together with title to make a complete sitelink. + :param title: Title of the page to associate. Use together with site to make a complete sitelink. + :param kwargs: More arguments for Python requests + :return: The answer from the Wikibase API + """ + params = { + 'action': 'wbeditentity', + 'data': ujson.dumps(data), + 'format': 'json' + } + + if baserevid: + params.update({'baserevid': baserevid}) + + if summary: + params.update({'summary': summary}) + + if tags: + params.update({'tags': '|'.join(tags)}) + + if not as_new: + if id: + params.update({'id': id}) + else: + assert site and title + params.update({ + 'site': site, + 'title': title + }) + + if clear: + params.update({'clear': ''}) + else: + assert type is not None + params.update({'new': type}) + + if is_bot: + params.update({'bot': ''}) + + return mediawiki_api_call_helper(data=params, login=login, is_bot=is_bot, **kwargs) + + def merge_items(from_id: str, to_id: str, login: Optional[_Login] = None, ignore_conflicts: Optional[List[str]] = None, is_bot: bool = False, **kwargs: Any) -> Dict: """ A static method to merge two items @@ -356,7 +418,8 @@ def remove_claims(claim_id: str, summary: Optional[str] = None, baserevid: Optio return mediawiki_api_call_helper(data=params, is_bot=is_bot, **kwargs) -def delete_page(title: Optional[str] = None, pageid: Optional[int] = None, reason: Optional[str] = None, deletetalk: bool = False, watchlist: str = 'preferences', watchlistexpiry: Optional[str] = None, +def delete_page(title: Optional[str] = None, pageid: Optional[int] = None, reason: Optional[str] = None, deletetalk: bool = False, watchlist: str = 'preferences', + watchlistexpiry: Optional[str] = None, login: Optional[_Login] = None, **kwargs: Any) -> Dict: """ Delete a page