update wechaty setting

Author: wj-Mcat

src/wechaty/plugin.py

@@ -33,7 +33,6 @@
 from datetime import datetime
 from telnetlib import Telnet
 import socket
-import json
 from typing import (
     TYPE_CHECKING,
     Iterable,
@@ -75,7 +74,7 @@
     error
 )
 from wechaty.types import EndPoint
-from wechaty.utils import HookDict
+from wechaty.utils import WechatySetting
 
 from .config import config
 
@@ -139,15 +138,17 @@ def _list_routes_txt(app: Quart) -> List[str]:
     rules = list(sorted(rules, key=lambda rule: rule.endpoint))
 
     headers = ("Endpoint", "Methods", "Websocket", "Rule")
-    rule_methods = [", ".join(sorted(rule.methods)) for rule in rules if rule.methods]
+    rule_methods = [", ".join(sorted(rule.methods))
+                    for rule in rules if rule.methods]
 
     widths = [
         max(len(rule.endpoint) for rule in rules),
         max(len(methods) for methods in rule_methods),
         len("Websocket"),
         max(len(rule.rule) for rule in rules),
     ]
-    widths = [max(len(header), width) for header, width in zip(headers, widths)]
+    widths = [max(len(header), width)
+              for header, width in zip(headers, widths)]
 
     # pylint: disable=C0209
     row = "{{0:<{0}}} | {{1:<{1}}} | {{2:<{2}}} | {{3:<{3}}}".format(*widths)
@@ -158,7 +159,8 @@ def _list_routes_txt(app: Quart) -> List[str]:
 
     for rule, methods in zip(rules, rule_methods):
         routes_txt.append(
-            row.format(rule.endpoint, methods, str(rule.websocket), rule.rule).rstrip()
+            row.format(rule.endpoint, methods, str(
+                rule.websocket), rule.rule).rstrip()
         )
     return routes_txt
 
@@ -175,7 +177,8 @@ def _signal_handler(*_: Any) -> None:  # noqa: N803
     for signal_name in ["SIGINT", "SIGTERM", "SIGBREAK"]:
         if hasattr(signal, signal_name):
             try:
-                loop.add_signal_handler(getattr(signal, signal_name), _signal_handler)
+                loop.add_signal_handler(
+                    getattr(signal, signal_name), _signal_handler)
             except NotImplementedError:
                 # Add signal handler may not be implemented on Windows
                 signal.signal(getattr(signal, signal_name), _signal_handler)
@@ -397,10 +400,9 @@ def __init__(self, options: Optional[WechatyPluginOptions] = None):
         self.options = options
         self._default_logger: Optional[Logger] = None
         self._cache_dir: Optional[str] = None
-        self.setting_file: str = os.path.join(self.cache_dir, 'setting.json')
 
-        if not os.path.exists(self.setting_file):
-            self.setting = {}   # type: ignore
+        self.setting_file: str = os.path.join(self.cache_dir, 'setting.json')
+        self._setting_wechaty_setting: WechatySetting = WechatySetting(self.setting_file)
 
     def metadata(self) -> NavMetadata:
         """get the default nav metadata
@@ -417,34 +419,27 @@ def metadata(self) -> NavMetadata:
         )
 
     @property
-    def setting(self) -> HookDict[str, Any]:
+    def setting(self) -> WechatySetting:
         """get the setting of a plugin"""
-        with open(self.setting_file, 'r', encoding='utf-8') as f:
-            setting = json.load(f)
-
-        def set_item_hooks(_: str, __: Any, value: dict) -> None:
-            """hook set_item event, and save the setting"""
-            self._save_setting(value)
-
-        return HookDict(setting, set_item_hooks=set_item_hooks)
+        return self._setting_wechaty_setting
 
-    def _save_setting(self, value: dict) -> None:
-        """update the plugin setting"""
-        with open(self.setting_file, 'w', encoding='utf-8') as f:
-            json.dump(value, f, ensure_ascii=False)
-
-    @setting.setter     # type: ignore
+    @setting.setter
     def setting(self, value: dict) -> None:
-        """update the plugin setting"""
-        self._save_setting(value)
+        """update setting with value
+
+        Args:
+            value (dict): the value of setting dict
+        """
+        self._setting_wechaty_setting.save_setting(value)
 
     def get_ui_dir(self) -> Optional[str]:
         """get the ui asset dir
         """
         # 1. get the customized ui dir according to the static UI_DIR attribute
         if self.UI_DIR:
             if os.path.exists(self.UI_DIR):
-                self.logger.info("finding the UI_DIR<%s> for plugin<%s>", self.UI_DIR, type(self))
+                self.logger.info(
+                    "finding the UI_DIR<%s> for plugin<%s>", self.UI_DIR, type(self))
                 return self.UI_DIR
 
         # 2. get the default uidir: ui/dist
@@ -454,7 +449,8 @@ def get_ui_dir(self) -> Optional[str]:
         plugin_dir = os.path.dirname(str(plugin_dir_path))
         ui_dir = os.path.join(plugin_dir, 'ui')
         if os.path.exists(ui_dir):
-            self.logger.info("finding the UI_DIR<%s> for plugin<%s>", ui_dir, type(self))
+            self.logger.info(
+                "finding the UI_DIR<%s> for plugin<%s>", ui_dir, type(self))
             return ui_dir
 
         self.logger.warning(
@@ -574,7 +570,8 @@ class WechatyPluginManager:     # pylint: disable=too-many-instance-attributes
     def __init__(
         self, wechaty: Wechaty,
         endpoint: EndPoint,
-        scheduler_options: Optional[Union[AsyncIOScheduler, WechatySchedulerOptions]] = None
+        scheduler_options: Optional[Union[AsyncIOScheduler,
+                                          WechatySchedulerOptions]] = None
     ):
         self._plugins: Dict[str, WechatyPlugin] = OrderedDict()
         self._wechaty: Wechaty = wechaty
@@ -591,9 +588,11 @@ def __init__(
             scheduler = AsyncIOScheduler()
 
             if isinstance(scheduler_options.job_store, str):
-                scheduler_options.job_store = SQLAlchemyJobStore(scheduler_options.job_store)
+                scheduler_options.job_store = SQLAlchemyJobStore(
+                    scheduler_options.job_store)
 
-            scheduler.add_jobstore(scheduler_options.job_store, scheduler_options.job_store_alias)
+            scheduler.add_jobstore(
+                scheduler_options.job_store, scheduler_options.job_store_alias)
         self.scheduler: AsyncIOScheduler = scheduler
 
         self.static_file_cacher = StaticFileCacher([
@@ -670,7 +669,7 @@ async def update_plugin_setting() -> Response:
                 return error(f'plugin<{name}> not exist ...')
 
             plugin: WechatyPlugin = self._plugins[name]
-            plugin.setting = data['setting']    # type: ignore
+            plugin.setting = data['setting']
             return success(None)
 
         @app.route("/js/<string:name>", methods=['GET'])
@@ -842,7 +841,8 @@ async def start(self) -> None:
         # re-fetch the routes info
         routes_txt = _list_routes_txt(self.app)
 
-        log.info('============================starting web service========================')
+        log.info(
+            '============================starting web service========================')
         log.info('starting web service at endpoint: <{%s}:{%d}>', host, port)
 
         shutdown_trigger = await get_shutdown_trigger()
@@ -861,7 +861,8 @@ async def start(self) -> None:
         for route_txt in routes_txt:
             log.info(route_txt)
 
-        log.info('============================web service has started========================')
+        log.info(
+            '============================web service has started========================')
 
     # pylint: disable=too-many-locals,too-many-statements,too-many-branches
     async def emit_events(self, event_name: str, *args: Any, **kwargs: Any) -> None:

src/wechaty/utils/__init__.py

@@ -2,13 +2,13 @@
 from .qr_code import qr_terminal
 from .type_check import default_str
 from .date_util import timestamp_to_date
-from .data_util import HookDict
+from .data_util import WechatySetting
 # from .type_check import type_check
 
 __all__ = [
     'qr_terminal',
     'default_str',
     'timestamp_to_date',
-    'HookDict'
+    'WechatySetting'
     # 'type_check'
 ]

src/wechaty/utils/data_util.py

@@ -19,59 +19,66 @@
 limitations under the License.
 """
 from __future__ import annotations
+import json
 
+import os
 from typing import (
-    Callable,
     Any,
-    Dict,
-    Iterable,
-    Iterator,
-    Optional,
-    Type,
-    TypeVar,
-    MutableMapping,
-    Generic
 )
+from collections import UserDict
 
-_KT = TypeVar("_KT")
-_VT = TypeVar("_VT")
 
-
-class HookDict(MutableMapping[_KT, _VT], Generic[_KT, _VT]):
-    """add hooks to the dict data
-    """
+class WechatySetting(UserDict):
+    """save setting into file when changed"""
     def __init__(
         self,
-        data: Dict[_KT, _VT],
-        set_item_hooks: Optional[Callable[[_KT, _VT, Dict[_KT, _VT]], None]] = None,
-        default_type: Type = int,
+        setting_file: str
     ):
-        self._dict_data = data
-        self._set_item_hooks = set_item_hooks
-        self.default_type = default_type
+        """init wechaty setting"""
+        super().__init__()
+        self.setting_file = setting_file
+        self._init_setting()
+        self.data = self.read_setting()
 
-    def __setitem__(self, key: _KT, value: Any) -> None:
-        """triggered by `data[key] = value`"""
-        self._dict_data[key] = value
-        if self._set_item_hooks is not None:
-            self._set_item_hooks(key, value, self._dict_data)
+    def _init_setting(self):
+        """init setting file"""
+        # 1. init setting dir
+        setting_dir = os.path.dirname(self.setting_file)
+        os.makedirs(setting_dir, exist_ok=True)
+
+        # 2. init setting file
+        if not os.path.exists(self.setting_file):
+            self.save_setting({})
 
-    def __getitem__(self, __k: _KT) -> _VT:
-        """get item data"""
-        return self._dict_data.get(__k, self.default_type())
-    
-    def __delitem__(self, __v: _KT) -> None:
-        """del item by key"""
-        del self._dict_data[__v]
-    
-    def __iter__(self) -> Iterator[_KT]:
-        """get the iterable data of data"""
-        return iter(self._dict_data)
-    
-    def __len__(self) -> int:
-        """get the length of dict data"""
-        return len(self._dict_data)
+        # 3. check the content of setting file
+        else:
+            with open(self.setting_file, 'r', encoding='utf-8') as f:
+                content = f.read().strip()
+            
+            if not content:
+                self.save_setting({})
+        
+    def read_setting(self) -> dict:
+        """read the setting from file
+
+        Returns:
+            dict: the data of setting file
+        """
+        with open(self.setting_file, 'r', encoding='utf-8') as f:
+            data = json.load(f)
+        return data
+        
+    def save_setting(self, value: dict) -> None:
+        """update the plugin setting"""
+        with open(self.setting_file, 'w', encoding='utf-8') as f:
+            json.dump(value, f, ensure_ascii=False)
+        self.data = value
+
+    def __setitem__(self, key: str, value: Any) -> None:
+        """triggered by `data[key] = value`"""
+        self.data[key] = value
+        self.save_setting(self.data)
 
     def to_dict(self) -> dict:
-        """get the source dict data"""
-        return self._dict_data
+        """return the dict data"""
+        return self.data

tests/plugin_test.py

@@ -2,7 +2,9 @@
 import json
 import os
 import tempfile
+import unittest
 from wechaty.plugin import WechatyPlugin
+from wechaty.utils.data_util import WechatySetting
 
 
 def test_setting():
@@ -11,7 +13,9 @@ def test_setting():
         plugin = WechatyPlugin()
 
         plugin.setting['unk'] = 11
-        plugin.setting['count'] += 20
+
+        assert 'count' not in plugin.setting
+        plugin.setting['count'] = 20
 
         # load the setting file
         assert os.path.exists(plugin.setting_file)
@@ -22,3 +26,47 @@ def test_setting():
         assert data['unk'] == 11
         assert data['count'] == 20
 
+class TestWechatySetting(unittest.TestCase):
+
+    def setUp(self) -> None:
+        self.tempdir = tempfile.TemporaryDirectory()
+    
+    def tearDown(self) -> None:
+        self.tempdir.cleanup()
+
+    
+    def test_simple_init(self):
+        setting_file = os.path.join(self.tempdir.name, 'simple_setting.json')
+        wechaty_setting: WechatySetting = WechatySetting(setting_file)
+
+        assert os.path.exists(setting_file)
+
+        wechaty_setting['a'] = 'a'
+        assert 'a' in wechaty_setting.read_setting()
+        assert wechaty_setting.read_setting()['a'] == 'a'
+        
+        assert 'b' not in wechaty_setting
+
+        assert wechaty_setting.get("b", "b") == "b"
+        
+        wechaty_setting.save_setting({"c": "c"})
+        assert 'a' not in wechaty_setting
+        assert 'c' in wechaty_setting
+    
+    def test_sub_setting(self):
+        setting_file = os.path.join(self.tempdir.name, "sub", 'simple_setting.json')
+        wechaty_setting: WechatySetting = WechatySetting(setting_file)
+
+        assert os.path.exists(setting_file)
+
+        wechaty_setting['a'] = 'a'
+        assert 'a' in wechaty_setting.read_setting()
+        assert wechaty_setting.read_setting()['a'] == 'a'
+        
+        assert 'b' not in wechaty_setting
+
+        assert wechaty_setting.get("b", "b") == "b"
+        
+        wechaty_setting.save_setting({"c": "c"})
+        assert 'a' not in wechaty_setting
+        assert 'c' in wechaty_setting