Add wbi_helpers.edit_entity()

Replace baseentity._write() with a call to wbi_helpers.edit_entity()

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:

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