refactoring

kvmd/apps/kvmd/api/export.py

@@ -21,7 +21,6 @@
 
 
 import asyncio
-import operator
 
 from typing import Any
 from typing import List
@@ -29,6 +28,8 @@ from typing import List
 from aiohttp.web import Request
 from aiohttp.web import Response
 
+from .... import tools
+
 from ....plugins.atx import BaseAtx
 
 from ..info import InfoManager
@@ -78,6 +79,6 @@ class ExportApi:
                 "",
             ])
         elif isinstance(value, dict):
-            for (sub_key, sub_value) in sorted(value.items(), key=operator.itemgetter(0)):
+            for (sub_key, sub_value) in tools.sorted_kvs(value):
                 sub_path = (f"{path}_{sub_key}" if sub_key != "parsed_flags" else path)
                 self.__append_prometheus_rows(rows, sub_value, sub_path)

kvmd/apps/kvmd/streamer.py

@@ -24,7 +24,6 @@ import signal
 import asyncio
 import asyncio.subprocess
 import dataclasses
-import operator
 
 from typing import Tuple
 from typing import List
@@ -37,6 +36,7 @@ import aiohttp
 
 from ...logging import get_logger
 
+from ... import tools
 from ... import aiotools
 from ... import aioproc
 from ... import htclient
@@ -315,7 +315,7 @@ class Streamer:  # pylint: disable=too-many-instance-attributes
                             mtime=float(response.headers["X-Timestamp"]),
                             headers=tuple(
                                 (key, value)
-                                for (key, value) in sorted(response.headers.items(), key=operator.itemgetter(0))
+                                for (key, value) in tools.sorted_kvs(response.headers)
                                 if key.lower().startswith("x-ustreamer-") or key.lower() in [
                                     "x-timestamp",
                                     "access-control-allow-origin",

kvmd/apps/kvmd/ugpio.py

@@ -21,7 +21,6 @@
 
 
 import asyncio
-import operator
 
 from typing import List
 from typing import Dict
@@ -38,6 +37,7 @@ from ...plugins.ugpio import GpioDriverOfflineError
 from ...plugins.ugpio import BaseUserGpioDriver
 from ...plugins.ugpio import get_ugpio_driver_class
 
+from ... import tools
 from ... import aiotools
 
 from ...yamlconf import Section
@@ -242,13 +242,13 @@ class UserGpio:
                 notifier=self.__notifier,
                 **drv_config._unpack(ignore=["instance_name", "notifier", "type"]),
             )
-            for (driver, drv_config) in sorted(config.drivers.items(), key=operator.itemgetter(0))
+            for (driver, drv_config) in tools.sorted_kvs(config.drivers)
         }
 
         self.__inputs: Dict[str, _GpioInput] = {}
         self.__outputs: Dict[str, _GpioOutput] = {}
 
-        for (channel, ch_config) in sorted(config.scheme.items(), key=operator.itemgetter(0)):
+        for (channel, ch_config) in tools.sorted_kvs(config.scheme):
             driver = self.__drivers[ch_config.driver]
             if ch_config.mode == "input":
                 self.__inputs[channel] = _GpioInput(channel, ch_config, driver)
@@ -281,14 +281,14 @@ class UserGpio:
 
     def sysprep(self) -> None:
         get_logger().info("Preparing User-GPIO drivers ...")
-        for (_, driver) in sorted(self.__drivers.items(), key=operator.itemgetter(0)):
+        for (_, driver) in tools.sorted_kvs(self.__drivers):
             driver.prepare()
 
     async def systask(self) -> None:
         get_logger(0).info("Running User-GPIO drivers ...")
         await asyncio.gather(*[
             driver.run()
-            for (_, driver) in sorted(self.__drivers.items(), key=operator.itemgetter(0))
+            for (_, driver) in tools.sorted_kvs(self.__drivers)
         ])
 
     async def cleanup(self) -> None:

kvmd/tools.py

@@ -20,10 +20,14 @@
 # ========================================================================== #
 
 
+import operator
 import functools
 
+from typing import Tuple
+from typing import List
 from typing import Dict
 from typing import Hashable
+from typing import TypeVar
 
 
 # =====
@@ -41,3 +45,11 @@ def rget(dct: Dict, *keys: Hashable) -> Dict:
     if not isinstance(result, dict):
         raise TypeError(f"Not a dict as result: {result!r} from {dct!r} at {list(keys)}")
     return result
+
+
+_DictKeyT = TypeVar("_DictKeyT")
+_DictValueT = TypeVar("_DictValueT")
+
+
+def sorted_kvs(dct: Dict[_DictKeyT, _DictValueT]) -> List[Tuple[_DictKeyT, _DictValueT]]:
+    return sorted(dct.items(), key=operator.itemgetter(0))

kvmd/yamlconf/dumper.py

@@ -21,13 +21,14 @@
 
 
 import textwrap
-import operator
 
 from typing import Generator
 from typing import Any
 
 import yaml
 
+from .. import tools
+
 from . import Section
 
 
@@ -37,7 +38,7 @@ def make_config_dump(config: Section, indent: int=4) -> str:
 
 
 def _inner_make_dump(config: Section, indent: int, _level: int=0) -> Generator[str, None, None]:
-    for (key, value) in sorted(config.items(), key=operator.itemgetter(0)):
+    for (key, value) in tools.sorted_kvs(config):
         if isinstance(value, Section):
             prefix = " " * indent * _level
             yield f"{prefix}{key}:"