onyx_online/vban-cmd-python
1
0
Fork 0
mirror of https://github.com/onyx-and-iris/vban-cmd-python.git synced 2025-11-04 07:21:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

deep_merge implemented

Browse Source 
recursively merges dicts in profiles

patch bump
This commit is contained in:
onyx-and-iris 2023-07-12 04:52:50 +01:00
parent 2a98707bf8
commit a4b91bf5c6
3 changed files with 24 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pyproject.toml
View File

@ -1,6 +1,6 @@
[tool.poetry] [tool.poetry]
name = "vban-cmd" name = "vban-cmd"
version = "2.3.0" version = "2.3.1"
description = "Python interface for the VBAN RT Packet Service (Sendtext)" description = "Python interface for the VBAN RT Packet Service (Sendtext)"
authors = ["onyx-and-iris <code@onyxandiris.online>"] authors = ["onyx-and-iris <code@onyxandiris.online>"]
license = "MIT" license = "MIT"

14
vban_cmd/util.py
View File

@ -73,4 +73,18 @@ def comp(t0: tuple, t1: tuple) -> Iterator[bool]:
yield True yield True
def deep_merge(dict1, dict2):
"""Generator function for deep merging two dicts"""
for k in set(dict1) | set(dict2):
if k in dict1 and k in dict2:
if isinstance(dict1[k], dict) and isinstance(dict2[k], dict):
yield k, dict(deep_merge(dict1[k], dict2[k]))
else:
yield k, dict2[k]
elif k in dict1:
yield k, dict1[k]
else:
yield k, dict2[k]
Socket = IntEnum("Socket", "register request response", start=0) Socket = IntEnum("Socket", "register request response", start=0)

14
vban_cmd/vbancmd.py
View File

@ -4,13 +4,13 @@ import time
from abc import ABCMeta, abstractmethod from abc import ABCMeta, abstractmethod
from pathlib import Path from pathlib import Path
from queue import Queue from queue import Queue
from typing import Iterable, Optional, Union from typing import Iterable, Union
from .error import VBANCMDError from .error import VBANCMDError
from .event import Event from .event import Event
from .packet import RequestHeader from .packet import RequestHeader
from .subject import Subject from .subject import Subject
from .util import Socket, script from .util import Socket, deep_merge, script
from .worker import Producer, Subscriber, Updater from .worker import Producer, Subscriber, Updater
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -194,14 +194,18 @@ class VbanCmd(metaclass=ABCMeta):
f"Known configs: {list(self.configs.keys())}", f"Known configs: {list(self.configs.keys())}",
) )
try: try:
config = self.configs[name].copy() config = self.configs[name]
except KeyError as e: except KeyError as e:
self.logger.error(("\n").join(ERR_MSG)) self.logger.error(("\n").join(ERR_MSG))
raise VBANCMDError(("\n").join(ERR_MSG)) from e raise VBANCMDError(("\n").join(ERR_MSG)) from e
if "extends" in config: if "extends" in config:
extended = config.pop("extends") extended = config["extends"]
config = self.configs[extended] | config config = {
k: v
for k, v in deep_merge(self.configs[extended], config)
if k not in ("extends")
}
self.logger.debug( self.logger.debug(
f"profile '{name}' extends '{extended}', profiles merged.." f"profile '{name}' extends '{extended}', profiles merged.."
) )