nodes/Lex-DRL-ComfyUI-DictTools
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #1 from Lex-DRL/dev

Browse Source 
Less strict input-dict validation
This commit is contained in:
Lex Darlog
2026-04-22 06:35:10 -03:00
committed by GitHub
parent ef71454870 af61aa220f
commit de7923cd1f
4 changed files with 32 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
__typing.py
+1 -1
View File
@@ -7,4 +7,4 @@ from typing import Any as _A, Optional as _O, Union as _U, TypeVar as _TypeVar
T = _TypeVar('T')
T2 = _TypeVar('T2')
FormatDict = _TypeVar('FormatDict', bound=_U[_t.Dict[str, _A], _t.Mapping[str, _A]])
DictMap = _TypeVar('DictMap', bound=_U[_t.Dict, _t.Mapping])
_dict_funcs.py
+15 -93
View File
@@ -1,119 +1,41 @@
# encoding: utf-8
"""
Shared code to deal with dicts.
Shared low-level utility functions for dicts.
"""
import typing as _t
from frozendict import frozendict as _frozendict
import re as _re
from frozendict import deepfreeze as _deepfreeze, frozendict as _frozendict
from .__typing import T as _T, T2 as _T2
from .__typing import _t, _A, _O, _U, DictMap as _DictMap
_re_valid_key_match = _re.compile("[a-zA-Z_][a-zA-Z_0-9]*$").match
_re_starting_digits_match = _re.compile("[0-9]+").match
def _validate_key(key: str, errors_dict: _t.Dict[_t.Any, str]) -> _t.Union[str, None]:
"""Verify that the given string can be a key. Otherwise, append a message to `errors_dict`."""
if not isinstance(key, str):
errors_dict[key] = f"Not a valid key (string name): {key!r}"
return None
key = str(key).strip()
if not key:
errors_dict[key] = f"Key (string name) can't be empty. Got: {key!r}"
return None
if not _re_valid_key_match(key):
if _re_starting_digits_match(key):
errors_dict[key] = f"Key (string name) can't start with a digit. Got: {key!r}"
else:
errors_dict[key] = f"Key (string name) must contain only numbers, latin letters and underscores. Got: {key!r}"
return None
return key
def _raise_from_errors_dict(
errors_dict: _t.Dict[_t.Any, str],
single_error_format: str = '{}',
multi_errors_format: str = "Invalid keys (string names): {}",
):
if not errors_dict:
return
if len(errors_dict) == 1:
msg = single_error_format.format(next(iter(errors_dict.values())))
else:
msg = multi_errors_format.format(', '.join(repr(x) for x in errors_dict.keys()))
raise KeyError(msg)
def _verify_input_dict(input_dict: _t.Dict[str, _T] = None, error_if_none=False):
def __verify_input_dict_into_new(input_dict: _O[_DictMap], error_if_none=False) -> dict:
"""
Verify input dict to have only valid keys. Raises errors if invalid ones found.
Verify input dict.
Always returns a new dict instance (shallow copy if non-empty dict passed).
"""
if input_dict is None:
if error_if_none:
raise TypeError("No input-dict")
return
return dict()
# In py3.10, frozendict isn't a dict, but is a `typing.Mapping`.
# So, this many types to check against:
if not isinstance(input_dict, (dict, _frozendict, _t.Mapping)):
raise TypeError(f"Input-dict isn't a dict. Got: {input_dict!r}")
errors_dict: _t.Dict[_t.Any, str] = dict()
for key in input_dict.keys():
_validate_key(key, errors_dict)
_raise_from_errors_dict(
errors_dict,
"Invalid input-dict: {}", "Invalid keys (string names) in input-dict: {}"
)
def _verify_input_dict_into_new(input_dict: _t.Dict[str, _T] = None) -> _t.Dict[str, _T]:
"""
Verify input dict to have only valid keys. Raises errors if invalid ones found.
Always returns a new dict instance (shallow copy if non-empty dict passed).
"""
if input_dict is None:
return dict()
_verify_input_dict(input_dict)
return dict(input_dict)
# ----------------------------------------------------------
def _new_updated_dict(
input_dict: _t.Union[_t.Dict[str, _T], None], *updating_dicts: _t.Dict[str, _T2],
sort=True, frozen=True, validate_new_keys=True
) -> _t.Dict[str, _t.Union[_T, _T2]]:
input_dict: _O[_DictMap], *updating_dicts: _DictMap,
sort=True, frozen=True
) -> _U[dict, _frozendict]:
"""Return a new dict being a union of the input one and the new (updating) ones."""
out_dict: _t.Dict[str, _t.Union[_T, _T2]] = _verify_input_dict_into_new(input_dict)
out_dict = __verify_input_dict_into_new(input_dict)
if validate_new_keys:
errors_dict: _t.Dict[_t.Any, str] = dict()
for upd_d in updating_dicts:
upd_d = dict(upd_d)
for key in list(upd_d.keys()):
val = upd_d.pop(key)
key = _validate_key(key, errors_dict)
upd_d[key] = val
out_dict.update(upd_d)
_raise_from_errors_dict(errors_dict)
else:
for upd_d in updating_dicts:
out_dict.update(upd_d)
for upd_d in updating_dicts:
out_dict.update(upd_d)
if sort:
out_dict = {k: v for k, v in sorted(out_dict.items())} # rely on implied ordered dicts in newer py3 versions
return _deepfreeze(out_dict) if frozen else out_dict
def _new_dict_with_updated_key(
input_dict: _t.Union[_t.Dict[str, _T], None], key: str, value: _T2, sort=True, frozen=True, validate_key=True
) -> _t.Dict[str, _t.Union[_T, _T2]]:
"""Return a new dict being a union of the input one and the new (updating) key/value."""
return _new_updated_dict(
input_dict, {key: value}, sort=sort, frozen=frozen, validate_new_keys=validate_key
)
return _frozendict(out_dict) if frozen else out_dict
node_dict_from_text.py
+7 -7
View File
@@ -3,13 +3,15 @@
from inspect import cleandoc as _cleandoc
from frozendict import frozendict as _frozendict
from comfy_api.latest import io as _io, ui as _ui
from .__meta import (
category as _category,
pack_id_suffix as _pack_id
)
from .__typing import _A, _U, _O, _t, T as _T, FormatDict as _FormatDict
from .__typing import _A, _U, _O, _t, T as _T, DictMap as _DictMap
from ._dict_funcs import _new_updated_dict
from ._io_custom import (
_BaseNode,
@@ -18,8 +20,6 @@ from ._io_custom import (
)
from .docstring_formatter import format_docstring as _format_docstring
_dict = dict
# ----------------------------------------------------------
def _return_line_raw(line_raw: str, line_stripped:str) -> str:
@@ -113,15 +113,15 @@ class DictFromText(_BaseNode):
@classmethod
def execute(cls,
cleanup: bool, dict_text: str, show_status: bool = False,
dict: _O[_FormatDict] = None
dict: _O[_DictMap] = None
) -> _io.NodeOutput:
parsed_items = _parsed_kv_pairs_gen(dict_text, strip_lines=cleanup)
new_dict = {k: v for k, v in parsed_items}
if not new_dict:
# No need to create another dict instance if we add nothing:
out_dict = _dict() if dict is None else dict
out_ui = '' if show_status else None
out_dict = _frozendict() if dict is None else dict
out_ui = _ui.PreviewText('') if show_status else None
return _io.NodeOutput(out_dict, ui=out_ui)
out_dict = _new_updated_dict(dict, new_dict)
@@ -151,7 +151,7 @@ class DictFromTextOld1(_BaseNode):
@classmethod
def execute(cls,
cleanup: bool, strings: str, show_status: bool = False,
dict: _O[_FormatDict] = None
dict: _O[_DictMap] = None
) -> _io.NodeOutput:
return DictFromText.execute(
cleanup=cleanup, dict_text=strings, show_status=show_status,
nodes_dict.py
+9 -9
View File
@@ -11,8 +11,8 @@ from .__meta import (
category as _category,
pack_id_suffix as _pack_id
)
from .__typing import _A, _U, _O, _t, T as _T, FormatDict as _FormatDict
from ._dict_funcs import _new_dict_with_updated_key
from .__typing import _A, _U, _O, _t, T as _T, DictMap as _DictMap
from ._dict_funcs import _new_updated_dict
from ._io_custom import (
_BaseNode,
_DICT_INPUT_OPTIONAL, _DICT_OUTPUT,
@@ -52,16 +52,16 @@ class DictAddAny(_BaseNode):
@classmethod
def execute(cls,
name: str, dict: _O[_FormatDict] = None, value: _A = None
name: _O[str], dict: _O[_DictMap] = None, value: _A = None
) -> _io.NodeOutput:
"""Update/append an item of any type to the dict."""
if value is None:
if name is None:
if dict is None:
dict = _frozendict()
# No need to create another dict instance if we add nothing:
result = dict
else:
result = _new_dict_with_updated_key(dict, name, value)
result = _new_updated_dict(dict, {name: value})
return _io.NodeOutput(result)
# ----------------------------------------------------------
@@ -105,7 +105,7 @@ class DictAddString(_BaseNode):
@classmethod
def execute(cls,
name: str, cleanup: bool, string: str,
dict: _O[_FormatDict] = None,
dict: _O[_DictMap] = None,
) -> _io.NodeOutput:
"""Update/append a string to the dict."""
string = '' if string is None else str(string)
@@ -113,7 +113,7 @@ class DictAddString(_BaseNode):
string = '\n'.join(
x.strip() for x in string.splitlines()
)
result= _new_dict_with_updated_key(dict, name, string)
result= _new_updated_dict(dict, {name: string})
return _io.NodeOutput(result)
# ----------------------------------------------------------
@@ -160,7 +160,7 @@ class DictExtractString(_BaseNode):
@classmethod
def execute(cls,
cleanup_key: bool, key: str, show_status: bool = False,
dict: _O[_FormatDict] = None
dict: _O[_DictMap] = None
) -> _io.NodeOutput:
"""Extract a single string from the Format-Dict."""
if cleanup_key:
@@ -224,7 +224,7 @@ class DictExtractStringOld1(_BaseNode):
@classmethod
def execute(cls,
name: str, show_status: bool = False,
dict: _O[_FormatDict] = None
dict: _O[_DictMap] = None
) -> _io.NodeOutput:
return DictExtractString.execute(
cleanup_key=False, key=name, show_status=show_status, dict=dict