Converted persistent history files from pickle to JSON format

Author: kmvanbrunt

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 2.1.0 (TBD, 2021)
+* Enhancements
+   * Converted persistent history files from pickle to JSON format
+    
 ## 2.0.1 (June 7, 2021)
 * Bug Fixes
   * Exclude `plugins` and `tests_isolated` directories from tarball published to PyPI for `cmd2` release

cmd2/cmd2.py

@@ -34,7 +34,6 @@
 import glob
 import inspect
 import os
-import pickle
 import pydoc
 import re
 import sys
@@ -4443,15 +4442,14 @@ def _get_history(self, args: argparse.Namespace) -> 'OrderedDict[int, HistoryIte
     def _initialize_history(self, hist_file: str) -> None:
         """Initialize history using history related attributes
 
-        This function can determine whether history is saved in the prior text-based
-        format (one line of input is stored as one line in the file), or the new-as-
-        of-version 0.9.13 pickle based format.
-
-        History created by versions <= 0.9.12 is in readline format, i.e. plain text files.
-
-        Initializing history does not effect history files on disk, versions >= 0.9.13 always
-        write history in the pickle format.
+        :param hist_file: optional path to persistent history file. If specified, then history from
+                          previous sessions will be included. Additionally, all history will be written
+                          to this file when the application exits.
         """
+        from json import (
+            JSONDecodeError,
+        )
+
         self.history = History()
         # with no persistent history, nothing else in this method is relevant
         if not hist_file:
@@ -4474,36 +4472,27 @@ def _initialize_history(self, hist_file: str) -> None:
             self.perror(f"Error creating persistent history file directory '{hist_file_dir}': {ex}")
             return
 
-        # first we try and unpickle the history file
-        history = History()
-
+        # Read and process history file
         try:
-            with open(hist_file, 'rb') as fobj:
-                history = pickle.load(fobj)
-        except (
-            AttributeError,
-            EOFError,
-            FileNotFoundError,
-            ImportError,
-            IndexError,
-            KeyError,
-            ValueError,
-            pickle.UnpicklingError,
-        ):
-            # If any of these errors occur when attempting to unpickle, just use an empty history
+            with open(hist_file, 'r') as fobj:
+                history_json = fobj.read()
+            self.history = History.from_json(history_json)
+        except FileNotFoundError:
+            # Just use an empty history
             pass
         except OSError as ex:
             self.perror(f"Cannot read persistent history file '{hist_file}': {ex}")
             return
+        except (JSONDecodeError, KeyError, ValueError) as ex:
+            self.perror(f"Error processing persistent history file '{hist_file}': {ex}")
 
-        self.history = history
         self.history.start_session()
         self.persistent_history_file = hist_file
 
         # populate readline history
         if rl_type != RlType.NONE:
             last = None
-            for item in history:
+            for item in self.history:
                 # Break the command into its individual lines
                 for line in item.raw.splitlines():
                     # readline only adds a single entry for multiple sequential identical lines
@@ -4520,14 +4509,14 @@ def _initialize_history(self, hist_file: str) -> None:
         atexit.register(self._persist_history)
 
     def _persist_history(self) -> None:
-        """Write history out to the history file"""
+        """Write history out to the persistent history file as JSON"""
         if not self.persistent_history_file:
             return
 
         self.history.truncate(self._persistent_history_length)
         try:
-            with open(self.persistent_history_file, 'wb') as fobj:
-                pickle.dump(self.history, fobj)
+            with open(self.persistent_history_file, 'w') as fobj:
+                fobj.write(self.history.to_json())
         except OSError as ex:
             self.perror(f"Cannot write persistent history file '{self.persistent_history_file}': {ex}")
 

cmd2/history.py

@@ -3,12 +3,15 @@
 History management classes
 """
 
+import json
 import re
 from collections import (
     OrderedDict,
 )
 from typing import (
+    Any,
     Callable,
+    Dict,
     Iterable,
     List,
     Optional,
@@ -33,6 +36,9 @@ class HistoryItem:
     _listformat = ' {:>4}  {}'
     _ex_listformat = ' {:>4}x {}'
 
+    # Used in JSON dictionaries
+    _statement_field = 'statement'
+
     statement: Statement = attr.ib(default=None, validator=attr.validators.instance_of(Statement))
 
     def __str__(self) -> str:
@@ -94,6 +100,22 @@ def pr(self, idx: int, script: bool = False, expanded: bool = False, verbose: bo
 
         return ret_str
 
+    def to_dict(self) -> Dict[str, Any]:
+        """Utility method to convert this HistoryItem into a dictionary for use in persistent JSON history files"""
+        return {HistoryItem._statement_field: self.statement.to_dict()}
+
+    @staticmethod
+    def from_dict(source_dict: Dict[str, Any]) -> 'HistoryItem':
+        """
+        Utility method to restore a HistoryItem from a dictionary
+
+        :param source_dict: source data dictionary (generated using to_dict())
+        :return: HistoryItem object
+        :raises KeyError: if source_dict is missing required elements
+        """
+        statement_dict = source_dict[HistoryItem._statement_field]
+        return HistoryItem(Statement.from_dict(statement_dict))
+
 
 class History(List[HistoryItem]):
     """A list of :class:`~cmd2.history.HistoryItem` objects with additional methods
@@ -109,6 +131,11 @@ class History(List[HistoryItem]):
     class to gain access to the historical record.
     """
 
+    # Used in JSON dictionaries
+    _history_version = '1.0.0'
+    _history_version_field = 'history_version'
+    _history_items_field = 'history_items'
+
     def __init__(self, seq: Iterable[HistoryItem] = ()) -> None:
         super(History, self).__init__(seq)
         self.session_start_index = 0
@@ -301,3 +328,36 @@ def _build_result_dictionary(
             if filter_func is None or filter_func(self[index]):
                 results[index + 1] = self[index]
         return results
+
+    def to_json(self) -> str:
+        """Utility method to convert this History into a JSON string for use in persistent history files"""
+        json_dict = {
+            History._history_version_field: History._history_version,
+            History._history_items_field: [hi.to_dict() for hi in self],
+        }
+        return json.dumps(json_dict, ensure_ascii=False, indent=2)
+
+    @staticmethod
+    def from_json(history_json: str) -> 'History':
+        """
+        Utility method to restore History from a JSON string
+
+        :param history_json: history data as JSON string (generated using to_json())
+        :return: History object
+        :raises json.JSONDecodeError: if passed invalid JSON string
+        :raises KeyError: if JSON is missing required elements
+        :raises ValueError: if history version in JSON isn't supported
+        """
+        json_dict = json.loads(history_json)
+        version = json_dict[History._history_version_field]
+        if version != History._history_version:
+            raise ValueError(
+                f"Unsupported history file version: {version}. This application uses version {History._history_version}."
+            )
+
+        items = json_dict[History._history_items_field]
+        history = History()
+        for hi_dict in items:
+            history.append(HistoryItem.from_dict(hi_dict))
+
+        return history

cmd2/parsing.py

@@ -147,6 +147,9 @@ class Statement(str):  # type: ignore[override]
     # if output was redirected, the destination file token (quotes preserved)
     output_to: str = attr.ib(default='', validator=attr.validators.instance_of(str))
 
+    # Used in JSON dictionaries
+    _args_field = 'args'
+
     def __new__(cls, value: object, *pos_args: Any, **kw_args: Any) -> 'Statement':
         """Create a new instance of Statement.
 
@@ -221,6 +224,31 @@ def argv(self) -> List[str]:
 
         return rtn
 
+    def to_dict(self) -> Dict[str, Any]:
+        """Utility method to convert this Statement into a dictionary for use in persistent JSON history files"""
+        return self.__dict__.copy()
+
+    @staticmethod
+    def from_dict(source_dict: Dict[str, Any]) -> 'Statement':
+        """
+        Utility method to restore a Statement from a dictionary
+
+        :param source_dict: source data dictionary (generated using to_dict())
+        :return: Statement object
+        :raises KeyError: if source_dict is missing required elements
+        """
+        # value needs to be passed as a positional argument. It corresponds to the args field.
+        try:
+            value = source_dict[Statement._args_field]
+        except KeyError as ex:
+            raise KeyError(f"Statement dictionary is missing {ex} field")
+
+        # Pass the rest at kwargs (minus args)
+        kwargs = source_dict.copy()
+        del kwargs[Statement._args_field]
+
+        return Statement(value, **kwargs)
+
 
 class StatementParser:
     """Parse user input as a string into discrete command components."""