diff --git a/__typing.py b/__typing.py index dec7f96..a310bd2 100644 --- a/__typing.py +++ b/__typing.py @@ -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]) diff --git a/_dict_funcs.py b/_dict_funcs.py index 92876eb..2d08ef0 100644 --- a/_dict_funcs.py +++ b/_dict_funcs.py @@ -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 diff --git a/node_dict_from_text.py b/node_dict_from_text.py index 0c83048..ef7c0f3 100644 --- a/node_dict_from_text.py +++ b/node_dict_from_text.py @@ -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, diff --git a/nodes_dict.py b/nodes_dict.py index 590f12c..769ab63 100644 --- a/nodes_dict.py +++ b/nodes_dict.py @@ -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