sanity refactors, more docs, more tests

Author: fastily

pwiki/dwrap.py

@@ -8,19 +8,16 @@
 class _DataEntry:
     """Base template class for more structured data returned by the API."""
 
-    def __init__(self, user: str = None, title: str = None, summary: str = None, timestamp: str = None):
+    def __init__(self, e: dict):
         """Initializer, creates a new DataEntry.
 
         Args:
-            user (str, optional): The user associated with this entry. `User:` prefix should be omitted. Defaults to None.
-            title (str, optional): The page title associated with this entry. Defaults to None.
-            summary (str, optional): The summary associated with this entry. Defaults to None.
-            timestamp (str, optional): The timestamp (in ISO 8601) the entry occured at. Defaults to None.
+            e (dict): A json object from the list in the response from the server.
         """
-        self.user: str = user
-        self.title: str = title
-        self.summary: str = summary
-        self.timestamp: datetime = datetime.fromisoformat(timestamp.replace("Z", "+00:00")) if timestamp else None
+        self.user: str = e.get("user")
+        self.title: str = e.get("title")
+        self.summary: str = e.get("comment")
+        self.timestamp: datetime = datetime.fromisoformat(ts.replace("Z", "+00:00")) if (ts := e.get("timestamp")) else None
 
     def __repr__(self) -> str:
         """Creates a str representation of this DataEntry.  Useful for debugging.
@@ -40,7 +37,7 @@ def __init__(self, e: dict):
         Args:
             e (dict): A json object from the `"contributions"` list in the response from the server.
         """
-        super().__init__(e.get("user"), e.get("title"), e.get("comment"), e.get("timestamp"))
+        super().__init__(e)
 
         self.is_page_create: bool = e.get("new")
         self.is_minor: bool = e.get("minor")
@@ -56,7 +53,7 @@ def __init__(self, e: dict):
         Args:
             e (dict): A json object from the `"imageinfo"` list in the response from the server.
         """
-        super().__init__(e.get("user"), summary=e.get("comment"), timestamp=e.get("timestamp"))
+        super().__init__(e)
 
         self.size: int = e.get("size")
         self.width: int = e.get("width")
@@ -74,7 +71,7 @@ def __init__(self, e: dict):
         Args:
             e (dict): A json object from the `"logevents"` list in the response from the server.
         """
-        super().__init__(e.get("user"), e.get("title"), e.get("comment"), e.get("timestamp"))
+        super().__init__(e)
 
         self.type: str = e.get("type")
         self.action: str = e.get("action")
@@ -90,7 +87,7 @@ def __init__(self, e: dict):
         Args:
             e (dict): A json object from the `"revisions"` list in the response from the server.
         """
-        super().__init__(e.get("user"), summary=e.get("comment"), timestamp=e.get("timestamp"))
+        super().__init__(e)
 
         self.text: str = mine_for(e, "slots", "main", "content")
 

pwiki/oquery.py

@@ -41,50 +41,6 @@ def _pair_titles_query(wiki: Wiki, id: str, pl: dict, titles: list[str], desc: s
 
         return out
 
-    # @staticmethod
-    # def _prop_cont_single(wiki: Wiki, title: str, template: QConstant, extra_pl: dict = None) -> list:
-    #     out = []
-
-    #     params = {**template.pl_with_limit(), "prop": template.name, "titles": title} | (extra_pl or {})
-
-    #     while True:
-    #         if not (response := query_and_validate(wiki, params, desc=f"peform a prop_cont_single query with '{template.name}'")):
-    #             raise OSError(f"Critical failure performing a _prop_cont_single query with {template.name}, cannot proceed")
-
-    #         if not ((l := mine_for(response, "query", "pages")) and template.name in (p := l[0])):
-    #             break
-
-    #         out += template.retrieve_results(p[template.name])
-
-    #         if not (cont := get_continue_params(response)):
-    #             break
-
-    #         params.update(cont)
-
-    #     return out
-
-    # @staticmethod
-    # def _list_cont(wiki: Wiki, template: QConstant, extra_pl: dict = None) -> list:
-    #     out = []
-
-    #     params = {**template.pl_with_limit(), "list": template.name} | (extra_pl or {})
-    #     while True:
-    #         if not (response := query_and_validate(wiki, params, desc=f"peform a list_cont query with '{template.name}'")):
-    #             return
-
-    #         if template.name not in (q := mine_for(response, "query")):
-    #             log.error("'%s' was not found in query result while doing a list_cont, something is very wrong.", template.name)
-    #             return
-
-    #         out += template.retrieve_results(q[template.name])
-
-    #         if not (cont := get_continue_params(response)):
-    #             break
-
-    #         params.update(cont)
-
-    #     return out
-
     @staticmethod
     def fetch_token(wiki: Wiki, login_token: bool = False) -> str:
         """Fetch a csrf or login token from the server.  By default, this method will retrieve a csrf token.

pwiki/utils.py

@@ -42,7 +42,7 @@ def mine_for(target: dict, *keys: str) -> Any:
         keys (str): The keys to follow.
 
     Returns:
-        Any: Whatever value is found at the end of following the specified keys.  None if nothing was not found.
+        Any: Whatever value is found at the end of following the specified keys.  `None` if nothing was not found.
     """
     try:
         for k in keys:

pwiki/waction.py

@@ -22,30 +22,21 @@ class WAction:
     """Collection of functions which can perform write actions on a Wiki"""
 
     @staticmethod
-    def _post_action(wiki: Wiki, action: str, form: dict = None, apply_token: bool = True, timeout: int = 15, extra_args: dict = None) -> dict:
-        """Convienence method, performs the actual POST of the action to the server.
+    def _action_and_validate(wiki: Wiki, action: str, form: dict = None, apply_token: bool = True, timeout: int = 15, success_vals: tuple = ("Success",), extra_args: dict = None) -> dict:
+        """Performs a `_post_action()` and checks the results for errors.  If there is an error, it will be logged accordingly.
 
         Args:
             wiki (Wiki): The Wiki object to use
-            action (str): The action to perform.
+            action (str): The id of the action to perform.
             form (dict, optional): The parameters to POST to the server, if applicable. Defaults to None.
             apply_token (bool, optional): Set `True` to also send the Wiki's csrf token in the POST. Defaults to True.
+            timeout (int, optional): The length of time (in seconds) to wait before marking the action as failed. Defaults to 15.
+            success_vals (tuple, optional): The keyword responses returned by the server which indicate a successful action.  Optional, set `None` to skip this check.  Defaults to ("Success",).
+            extra_args (dict, optional): Any `kwargs` that should be passed to the underlying requests Session object when performing a POST. Defaults to None.
 
         Returns:
-            dict: The response from the server.  Empty dict if there was an error.
+            dict: The json response from the server, or `None` if something went wrong.
         """
-        pl = make_params(action, form) | ({"token": wiki.csrf_token} if apply_token else {})
-
-        try:
-            return wiki.client.post(wiki.endpoint, data=pl, **({"timeout": timeout} | (extra_args or {}))).json()
-        except Exception:
-            log.error("%s: Could not reach server or read response while performing %s with params %s", wiki, action, pl, exc_info=True)
-
-        return {}
-
-    @staticmethod
-    def _action_and_validate(wiki: Wiki, action: str, form: dict = None, apply_token: bool = True, timeout: int = 15, success_vals: tuple = ("Success",), extra_args: dict = None) -> dict:
-
         if not (response := WAction._post_action(wiki, action, form, apply_token, timeout, extra_args)):
             log.error("%s: No response from server while trying to perform action '%s'", wiki, action)
             log.debug("Sent parameters: %s", form)
@@ -56,12 +47,36 @@ def _action_and_validate(wiki: Wiki, action: str, form: dict = None, apply_token
             log.debug(response)
             return
 
-        if (status := mine_for(response, action, "result")) in success_vals:
+        if not success_vals or (status := mine_for(response, action, "result")) in success_vals:
             return response
 
         log.error("%s: Failed to perform action '%s', got bad result from server: %s", wiki, action, status)
         log.debug(response)
 
+    @staticmethod
+    def _post_action(wiki: Wiki, action: str, form: dict = None, apply_token: bool = True, timeout: int = 15, extra_args: dict = None) -> dict:
+        """Convienence method, performs the actual POST of the action to the server.
+
+        Args:
+            wiki (Wiki): The Wiki object to use
+            action (str): The action to perform.
+            form (dict, optional): The parameters to POST to the server, if applicable. Defaults to None.
+            apply_token (bool, optional): Set `True` to also send the Wiki's csrf token in the POST. Defaults to True.
+            timeout (int, optional): The length of time (in seconds) to wait before marking the action as failed. Defaults to 15.
+            extra_args (dict, optional): Any `kwargs` that should be passed to the underlying requests Session object when performing a POST. Defaults to None.
+
+        Returns:
+            dict: The response from the server.  Empty dict if there was an error.
+        """
+        pl = make_params(action, form) | ({"token": wiki.csrf_token} if apply_token else {})
+
+        try:
+            return wiki.client.post(wiki.endpoint, data=pl, **({"timeout": timeout} | (extra_args or {}))).json()
+        except Exception:
+            log.error("%s: Could not reach server or read response while performing %s with params %s", wiki, action, pl, exc_info=True)
+
+        return {}
+
     @staticmethod
     def delete(wiki: Wiki, title: str, reason: str) -> bool:
         """Deletes a page.  PRECONDITION: `wiki` must be logged in and have the ability to delete pages for this to work.
@@ -74,7 +89,7 @@ def delete(wiki: Wiki, title: str, reason: str) -> bool:
         Returns:
             bool: `True` if this action succeeded.
         """
-        return bool(WAction._action_and_validate(wiki, "delete", {"title": title, "reason": reason}))
+        return bool(WAction._action_and_validate(wiki, "delete", {"title": title, "reason": reason}, success_vals=None))
 
     @staticmethod
     def edit(wiki: Wiki, title: str, text: str = None, summary: str = "", prepend: str = None, append: str = None, minor: bool = False) -> bool:
@@ -83,10 +98,10 @@ def edit(wiki: Wiki, title: str, text: str = None, summary: str = "", prepend: s
         Args:
             wiki (Wiki): The Wiki to use.
             title (str): The title to edit.
-            text (str, optional): Text to replace the current page's contents with. Mutually exclusive with `prepend`/`append`. Defaults to None.
+            text (str, optional): Text to replace the current page's contents with.  Overrides `prepend`/`append`.  Defaults to None.
             summary (str, optional): The edit summary to use. Defaults to "".
-            prepend (str, optional): Text to prepend to the page. Mutually exclusive with `text`. Defaults to None.
-            append (str, optional): Text to append to the page.  Mutually exclusive with `text`. Defaults to None.
+            prepend (str, optional): Text to prepend to the page. Defaults to None.
+            append (str, optional): Text to append to the page. Defaults to None.
             minor (bool, optional): Set `True` to mark this edit as minor. Defaults to False.
 
         Raises: