✨ feat: config serializer #1 (#2)

Author: mic1on

pyproject.toml

@@ -17,6 +17,7 @@ packages = [
 [tool.poetry.dependencies]
 python = "^3.8"
 httpx = "^0.25.2"
+tomli = { version = "^2.0.0", python = "<3.11" }
 
 
 [tool.poetry.group.test.dependencies]

src/use_nacos/endpoints/config.py

@@ -1,15 +1,15 @@
 import asyncio
 import hashlib
-import json
 import logging
 import threading
-from typing import Optional, Any, Callable
+from typing import Optional, Any, Callable, Union
 
 import httpx
 
 from .endpoint import Endpoint
 from ..cache import BaseCache, MemoryCache, memory_cache
 from ..exception import HTTPResponseError
+from ..serializer import Serializer, AutoSerializer
 from ..typings import SyncAsync
 
 logger = logging.getLogger(__name__)
@@ -29,15 +29,25 @@ def _parse_config_key(key: str):
     return key.split('#')
 
 
+def _serialize_config(
+        config: Any,
+        serializer: Optional[Union["Serializer", bool]] = None
+):
+    """ Serialize config with serializer """
+    if isinstance(serializer, bool) and serializer is True:
+        serializer = AutoSerializer()
+    if isinstance(serializer, Serializer):
+        return serializer(config)
+    return config
+
+
 class _BaseConfigEndpoint(Endpoint):
 
     def _get(
             self,
             data_id: str,
             group: str,
-            tenant: Optional[str] = '',
-            *,
-            serialized: Optional[bool] = False
+            tenant: Optional[str] = ''
     ) -> SyncAsync[Any]:
         return self.client.request(
             "/nacos/v1/cs/configs",
@@ -46,7 +56,7 @@ def _get(
                 "group": group,
                 "tenant": tenant,
             },
-            serialized=serialized
+            serialized=False
         )
 
     def publish(
@@ -121,12 +131,10 @@ def subscriber(
 class ConfigOperationMixin:
 
     @staticmethod
-    def _config_callback(callback, config, serialized):
+    def _config_callback(callback, config, serializer):
         if not callable(callback):
             return
-
-        if serialized:
-            config = json.loads(config)
+        config = _serialize_config(config, serializer)
         callback(config)
 
     def get(
@@ -135,20 +143,20 @@ def get(
             group: str,
             tenant: Optional[str] = '',
             *,
-            serialized: Optional[bool] = False,
+            serializer: Optional[Union["Serializer", bool]] = None,
             cache: Optional[BaseCache] = None,
             default: Optional[str] = None
     ) -> SyncAsync[Any]:
         cache = cache or memory_cache
         config_key = _get_config_key(data_id, group, tenant)
         try:
-            config = self._get(data_id, group, tenant, serialized=serialized)
+            config = self._get(data_id, group, tenant)
             # todo: this function need to be optimized
             cache.set(config_key, config)
-            return config
+            return _serialize_config(config, serializer)
         except (httpx.ConnectError, httpx.TimeoutException) as exc:
             logger.error("Failed to get config from server, try to get from cache. %s", exc)
-            return cache.get(config_key)
+            return _serialize_config(cache.get(config_key), serializer)
         except HTTPResponseError as exc:
             logger.debug("Failed to get config from server. %s", exc)
             if exc.status == 404 and default is not None:
@@ -161,7 +169,7 @@ def subscribe(
             group: str,
             tenant: Optional[str] = '',
             timeout: Optional[int] = 30_000,
-            serialized: Optional[bool] = False,
+            serializer: Optional[Union["Serializer", bool]] = None,
             cache: Optional[BaseCache] = None,
             callback: Optional[Callable] = None
     ) -> SyncAsync[Any]:
@@ -179,10 +187,10 @@ def _subscriber():
                     if not response:
                         continue
                     logging.info("Configuration update detected.")
-                    last_config = self.get(data_id, group, tenant, serialized=False)
+                    last_config = self._get(data_id, group, tenant)
                     last_md5 = _get_md5(last_config)
                     cache.set(config_key, last_config)
-                    self._config_callback(callback, last_config, serialized)
+                    self._config_callback(callback, last_config, serializer)
                 except Exception as exc:
                     logging.error(exc)
                     stop_event.wait(1)
@@ -195,12 +203,11 @@ def _subscriber():
 class ConfigAsyncOperationMixin:
 
     @staticmethod
-    async def _config_callback(callback, config, serialized):
+    async def _config_callback(callback, config, serializer):
         if not callable(callback):
             return
 
-        if serialized:
-            config = json.loads(config)
+        config = _serialize_config(config, serializer)
         if asyncio.iscoroutinefunction(callback):
             await callback(config)
         else:
@@ -212,19 +219,19 @@ async def get(
             group: str,
             tenant: Optional[str] = '',
             *,
-            serialized: Optional[bool] = False,
+            serializer: Optional[Union["Serializer", bool]] = None,
             cache: Optional[BaseCache] = None,
             default: Optional[str] = None
     ) -> SyncAsync[Any]:
         cache = cache or memory_cache
         config_key = _get_config_key(data_id, group, tenant)
         try:
-            config = await self._get(data_id, group, tenant, serialized=serialized)
+            config = await self._get(data_id, group, tenant)
             cache.set(config_key, config)
-            return config
+            return _serialize_config(config, serializer)
         except (httpx.ConnectError, httpx.TimeoutException) as exc:
             logger.error("Failed to get config from server, try to get from cache. %s", exc)
-            return cache.get(config_key)
+            return _serialize_config(cache.get(config_key), serializer)
         except HTTPResponseError as exc:
             logger.debug("Failed to get config from server. %s", exc)
             if exc.status == 404 and default is not None:
@@ -237,7 +244,7 @@ async def subscribe(
             group: str,
             tenant: Optional[str] = '',
             timeout: Optional[int] = 30_000,
-            serialized: Optional[bool] = False,
+            serializer: Optional[Union["Serializer", bool]] = None,
             cache: Optional[BaseCache] = None,
             callback: Optional[Callable] = None,
     ) -> SyncAsync[Any]:
@@ -257,10 +264,10 @@ async def _async_subscriber():
                     if not response:
                         continue
                     logging.info("Configuration update detected.")
-                    last_config = await self.get(data_id, group, tenant, serialized=False)
+                    last_config = await self._get(data_id, group, tenant)
                     last_md5 = _get_md5(last_config)
                     cache.set(config_key, last_config)
-                    await self._config_callback(callback, last_config, serialized)
+                    await self._config_callback(callback, last_config, serializer)
                 except asyncio.CancelledError:
                     break
                 except Exception as exc:

src/use_nacos/serializer.py

@@ -0,0 +1,116 @@
+import abc
+import json
+import sys
+
+import yaml
+
+if sys.version_info >= (3, 11):
+    import tomllib
+else:
+    import tomli as tomllib
+
+
+class Serializer(abc.ABC):
+
+    @abc.abstractmethod
+    def __call__(self, *args, **kwargs):
+        raise NotImplementedError
+
+
+class TextSerializer(Serializer):
+    """
+    >>> text = TextSerializer()
+    >>> text('a = 1')
+    'a = 1'
+    >>> text('a = 1\\n[foo]\\nb = 2')
+    'a = 1\\n[foo]\\nb = 2'
+    """
+
+    def __call__(self, data) -> str:
+        return data
+
+
+class JsonSerializer(Serializer):
+    """
+    >>> json_ = JsonSerializer()
+    >>> json_('{"a": 1}')
+    {'a': 1}
+    >>> json_('{"a": 1, "foo": {"b": 2}}')
+    {'a': 1, 'foo': {'b': 2}}
+    """
+
+    def __call__(self, data) -> dict:
+        try:
+            return json.loads(data)
+        except json.JSONDecodeError:
+            raise SerializerException(f"Cannot parse data: {data!r}")
+
+
+class YamlSerializer(Serializer):
+    """
+    >>> yaml_ = YamlSerializer()
+    >>> yaml_('a: 1')
+    {'a': 1}
+    >>> yaml_('a: 1\\nfoo:\\n  b: 2')
+    {'a': 1, 'foo': {'b': 2}}
+    """
+
+    def __call__(self, data) -> dict:
+        try:
+            return yaml.safe_load(data)
+        except yaml.YAMLError:
+            raise SerializerException(f"Cannot parse data: {data!r}")
+
+
+class TomlSerializer(Serializer):
+    """
+    >>> toml = TomlSerializer()
+    >>> toml('a = 1')
+    {'a': 1}
+    >>> toml('a = 1\\n[foo]\\nb = 2')
+    {'a': 1, 'foo': {'b': 2}}
+    """
+
+    def __call__(self, data) -> dict:
+        try:
+            return tomllib.loads(data)
+        except Exception:
+            raise SerializerException(f"Cannot parse data: {data!r}")
+
+
+class SerializerException(Exception):
+    pass
+
+
+class AutoSerializer(Serializer):
+    """
+    >>> auto = AutoSerializer()
+    >>> auto('a = 1')
+    {'a': 1}
+    >>> auto('a = 1\\n[foo]\\nb = 2')
+    {'a': 1, 'foo': {'b': 2}}
+    >>> auto('{"a": 1}')
+    {'a': 1}
+    >>> auto('{"a": 1, "foo": {"b": 2}}')
+    {'a': 1, 'foo': {'b': 2}}
+    >>> auto('a: 1')
+    {'a': 1}
+    >>> auto('a: 1\\nfoo:\\n  b: 2')
+    {'a': 1, 'foo': {'b': 2}}
+    """
+
+    def __init__(self):
+        self.serializers = (
+            JsonSerializer(),
+            TomlSerializer(),
+            YamlSerializer(),
+            TextSerializer(),
+        )
+
+    def __call__(self, data) -> dict:
+        for serializer in self.serializers:
+            try:
+                return serializer(data)
+            except:
+                pass
+        raise SerializerException(f"Cannot parse data: {data!r}")

tests/test_config.py

@@ -8,8 +8,9 @@
 
 from use_nacos.cache import MemoryCache
 from use_nacos.client import NacosClient, NacosAsyncClient
-from use_nacos.endpoints import ConfigEndpoint, config as conf
+from use_nacos.endpoints import ConfigEndpoint, ConfigAsyncEndpoint, config as conf
 from use_nacos.exception import HTTPResponseError
+from use_nacos.serializer import JsonSerializer, AutoSerializer, YamlSerializer, TomlSerializer
 
 server_addr = os.environ.get('SERVER_ADDR')
 
@@ -31,7 +32,7 @@ def config(client):
 
 @pytest.fixture
 def async_config(async_client):
-    yield ConfigEndpoint(async_client)
+    yield ConfigAsyncEndpoint(async_client)
 
 
 def test_config_get_not_found(config):
@@ -48,7 +49,7 @@ def test_config_get_not_found_default_value(config, mocker):
     ('test_config', 'DEFAULT_GROUP'),
 ])
 @pytest.mark.parametrize('content ,tenant, type, serialized, expected', [
-    ('test_config', '', None, False, 'test_config'),
+    ('test_config', '', None, None, 'test_config'),
     (json.dumps({"a": "b"}), '', 'json', True, {"a": "b"}),
     ('<p>hello nacos</p>', '', 'html', False, '<p>hello nacos</p>'),
     (1234, '', None, True, 1234),
@@ -59,7 +60,7 @@ def test_config_publish_get(config, data_id, group, content, tenant, type, seria
         data_id,
         group,
         tenant,
-        serialized=serialized
+        serializer=serialized
     ) == expected
 
 
@@ -101,20 +102,20 @@ async def test_async_config_get_not_found(async_config):
 @pytest.mark.parametrize('data_id, group', [
     ('test_config', 'DEFAULT_GROUP'),
 ])
-@pytest.mark.parametrize('content ,tenant, type, serialized, expected', [
+@pytest.mark.parametrize('content ,tenant, type, serializer, expected', [
     ('test_config', '', None, False, 'test_config'),
-    (json.dumps({"a": "b"}), '', 'json', True, {"a": "b"}),
+    (json.dumps({"a": "b"}), '', 'json', JsonSerializer(), {"a": "b"}),
     ('<p>hello nacos</p>', '', 'html', False, '<p>hello nacos</p>'),
     (1234, '', None, True, 1234),
 ])
 @pytest.mark.asyncio
-async def test_async_config_publish_get(async_config, data_id, group, content, tenant, type, serialized, expected):
+async def test_async_config_publish_get(async_config, data_id, group, content, tenant, type, serializer, expected):
     assert await async_config.publish(data_id, group, content, tenant, type)
     assert await async_config.get(
         data_id,
         group,
         tenant,
-        serialized=serialized
+        serializer=serializer
     ) == expected
 
 
@@ -165,3 +166,13 @@ def test_config_from_cache(config, mocker):
     # mock timeout
     mocker.patch.object(ConfigEndpoint, '_get', side_effect=httpx.TimeoutException(""))
     assert config.get('test_config_cache', 'DEFAULT_GROUP', cache=mc) == "abc"
+
+
+@pytest.mark.parametrize("conf_str, serializer, expected", [
+    ("123", AutoSerializer(), 123),
+    ('{"a": 2}', JsonSerializer(), {"a": 2}),
+    ('a: 1\nfoo:\n  b: 2', YamlSerializer(), {'a': 1, 'foo': {'b': 2}}),
+    ('a = 1\n[foo]\nb = 2', TomlSerializer(), {'a': 1, 'foo': {'b': 2}}),
+])
+def test_config_serializer(conf_str, serializer, expected):
+    assert conf._serialize_config(conf_str, serializer) == expected