16820655284

Committed 07 Aug 2025 05:03PM UTC coverage: 86.841% (-0.05%) from 86.892%

Build # 16820655284

Build Type

push

github

Committed by

web-flow

Commit Message

CFNV2: support CDK bootstrap and deployment (#12967)

Coverage Stats

32 of 38 new or added lines in 5 files covered. (84.21%)

2013 existing lines in 125 files now uncovered.

66606 of 76699 relevant lines covered (86.84%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

79.52

/localstack-core/localstack/utils/collections.py

"""
This package provides custom collection types, as well as tools to analyze
and manipulate python collection (dicts, list, sets).
"""

import logging
import re
from collections.abc import Iterable, Iterator, Mapping, Sized
from typing import (
    Any,
    Callable,
    Optional,
    TypedDict,
    TypeVar,
    Union,
    cast,
    get_args,
    get_origin,
)

import cachetools

LOG = logging.getLogger(__name__)

# default regex to match an item in a comma-separated list string
DEFAULT_REGEX_LIST_ITEM = r"[\w-]+"


class AccessTrackingDict(dict):
    """
    Simple utility class that can be used to track (write) accesses to a dict's attributes.
    Note: could also be written as a proxy, to preserve the identity of "wrapped" - for now, it
          simply duplicates the entries of "wrapped" in the constructor, for simplicity.
    """

    def __init__(self, wrapped, callback: Callable[[dict, str, list, dict], Any] = None):
        super().__init__(wrapped)
        self.callback = callback

    def __setitem__(self, key, value):
        self.callback and self.callback(self, "__setitem__", [key, value], {})
        return super().__setitem__(key, value)


class DelSafeDict(dict):
    """Useful when applying jsonpatch. Use it as follows:

    obj.__dict__ = DelSafeDict(obj.__dict__)
    apply_patch(obj.__dict__, patch)
    """

    def __delitem__(self, key, *args, **kwargs):
        self[key] = None


class ImmutableList(tuple):
    """
    Wrapper class to create an immutable view of a given list or sequence.
    Note: Currently, this is simply a wrapper around `tuple` - could be replaced with
    custom implementations over time, if needed.
    """


class HashableList(ImmutableList):
    """Hashable, immutable list wrapper that can be used with dicts or hash sets."""

    def __hash__(self):
        return sum(hash(i) for i in self)


class ImmutableDict(Mapping):
    """Wrapper class to create an immutable view of a given list or sequence."""

    def __init__(self, seq=None, **kwargs):
        self._dict = dict(seq, **kwargs)

    def __len__(self) -> int:
        return self._dict.__len__()

    def __iter__(self) -> Iterator:
        return self._dict.__iter__()

    def __getitem__(self, key):
        return self._dict.__getitem__(key)

    def __eq__(self, other):
        return self._dict.__eq__(other._dict if isinstance(other, ImmutableDict) else other)

    def __str__(self):
        return self._dict.__str__()


class HashableJsonDict(ImmutableDict):
    """
    Simple dict wrapper that can be used with dicts or hash sets. Note: the assumption is that the dict
    can be JSON-encoded (i.e., must be acyclic and contain only lists/dicts and simple types)
    """

    def __hash__(self):
        from localstack.utils.json import canonical_json

        return hash(canonical_json(self._dict))


_ListType = TypeVar("_ListType")


class PaginatedList(list[_ListType]):
    """List which can be paginated and filtered. For usage in AWS APIs with paginated responses"""

    DEFAULT_PAGE_SIZE = 50

    def get_page(
        self,
        token_generator: Callable[[_ListType], str],
        next_token: str = None,
        page_size: int = None,
        filter_function: Callable[[_ListType], bool] = None,
    ) -> tuple[list[_ListType], Optional[str]]:
        if filter_function is not None:
            result_list = list(filter(filter_function, self))
        else:
            result_list = self

        if page_size is None:
            page_size = self.DEFAULT_PAGE_SIZE

        # returns all or remaining elements in final page.
        if len(result_list) <= page_size and next_token is None:
            return result_list, None

        start_idx = 0

        try:
            start_item = next(item for item in result_list if token_generator(item) == next_token)
            start_idx = result_list.index(start_item)
        except StopIteration:
            pass

        if start_idx + page_size < len(result_list):
            next_token = token_generator(result_list[start_idx + page_size])
        else:
            next_token = None

        return result_list[start_idx : start_idx + page_size], next_token


class CustomExpiryTTLCache(cachetools.TTLCache):
    """TTLCache that allows to set custom expiry times for individual keys."""

    def set_expiry(self, key: Any, ttl: Union[float, int]) -> float:
        """Set the expiry of the given key in a TTLCache to (<current_time> + <ttl>)"""
        with self.timer as time:
            # note: need to access the internal dunder API here
            self._TTLCache__getlink(key).expires = expiry = time + ttl
            return expiry


def get_safe(dictionary, path, default_value=None):
    """
    Performs a safe navigation on a Dictionary object and
    returns the result or default value (if specified).
    The function follows a common AWS path resolution pattern "$.a.b.c".

    :type dictionary: dict
    :param dictionary: Dict to perform safe navigation.

    :type path: list|str
    :param path: List or dot-separated string containing the path of an attribute,
                 starting from the root node "$".

    :type default_value: any
    :param default_value: Default value to return in case resolved value is None.

    :rtype: any
    :return: Resolved value or default_value.
    """
    if not isinstance(dictionary, dict) or len(dictionary) == 0:
        return default_value

    attribute_path = path if isinstance(path, list) else path.split(".")
    if len(attribute_path) == 0 or attribute_path[0] != "$":
        raise AttributeError('Safe navigation must begin with a root node "$"')

    current_value = dictionary
    for path_node in attribute_path:
        if path_node == "$":
            continue

        if re.compile("^\\d+$").search(str(path_node)):
            path_node = int(path_node)

        if isinstance(current_value, dict) and path_node in current_value:
            current_value = current_value[path_node]
        elif isinstance(current_value, list) and path_node < len(current_value):
            current_value = current_value[path_node]
        else:
            current_value = None

    return current_value or default_value


def set_safe_mutable(dictionary, path, value):
    """
    Mutates original dict and sets the specified value under provided path.

    :type dictionary: dict
    :param dictionary: Dict to mutate.

    :type path: list|str
    :param path: List or dot-separated string containing the path of an attribute,
                 starting from the root node "$".

    :type value: any
    :param value: Value to set under specified path.

    :rtype: dict
    :return: Returns mutated dictionary.
    """
    if not isinstance(dictionary, dict):
        raise AttributeError('"dictionary" must be of type "dict"')

    attribute_path = path if isinstance(path, list) else path.split(".")
    attribute_path_len = len(attribute_path)

    if attribute_path_len == 0 or attribute_path[0] != "$":
        raise AttributeError('Dict navigation must begin with a root node "$"')

    current_pointer = dictionary
    for i in range(attribute_path_len):
        path_node = attribute_path[i]

        if path_node == "$":
            continue

        if i < attribute_path_len - 1:
            if path_node not in current_pointer:
                current_pointer[path_node] = {}
            if not isinstance(current_pointer, dict):
                raise RuntimeError(
                    'Error while deeply setting a dict value. Supplied path is not of type "dict"'
                )
        else:
            current_pointer[path_node] = value

        current_pointer = current_pointer[path_node]

    return dictionary


def pick_attributes(dictionary, paths):
    """
    Picks selected attributes a returns them as a new dictionary.
    This function works as a whitelist of attributes to keep in a new dictionary.

    :type dictionary: dict
    :param dictionary: Dict to pick attributes from.

    :type paths: list of (list or str)
    :param paths: List of lists or strings with dot-separated paths, starting from the root node "$".

    :rtype: dict
    :return: Returns whitelisted dictionary.
    """
    new_dictionary = {}

    for path in paths:
        value = get_safe(dictionary, path)

        if value is not None:
            set_safe_mutable(new_dictionary, path, value)

    return new_dictionary


def select_attributes(obj: dict, attributes: list[str]) -> dict:
    """Select a subset of attributes from the given dict (returns a copy)"""
    attributes = attributes if is_list_or_tuple(attributes) else [attributes]
    return {k: v for k, v in obj.items() if k in attributes}


def remove_attributes(obj: dict, attributes: list[str], recursive: bool = False) -> dict:
    """Remove a set of attributes from the given dict (in-place)"""
    from localstack.utils.objects import recurse_object

    if recursive:

        def _remove(o, **kwargs):
            if isinstance(o, dict):
                remove_attributes(o, attributes)
            return o

        return recurse_object(obj, _remove)

    attributes = ensure_list(attributes)
    for attr in attributes:
        obj.pop(attr, None)
    return obj


def rename_attributes(
    obj: dict, old_to_new_attributes: dict[str, str], in_place: bool = False
) -> dict:
    """Rename a set of attributes in the given dict object. Second parameter is a dict that maps old to
    new attribute names. Default is to return a copy, but can also pass in_place=True."""
    if not in_place:
        obj = dict(obj)
    for old_name, new_name in old_to_new_attributes.items():
        if old_name in obj:
            obj[new_name] = obj.pop(old_name)
    return obj


def is_list_or_tuple(obj) -> bool:
    return isinstance(obj, (list, tuple))


def ensure_list(obj: Any, wrap_none=False) -> Optional[list]:
    """Wrap the given object in a list, or return the object itself if it already is a list."""
    if obj is None and not wrap_none:
        return obj
    return obj if isinstance(obj, list) else [obj]


def to_unique_items_list(inputs, comparator=None):
    """Return a list of unique items from the given input iterable.
    The comparator(item1, item2) returns True/False or an int for comparison."""

    def contained(item):
        for r in result:
            if comparator:
                cmp_res = comparator(item, r)
                if cmp_res is True or str(cmp_res) == "0":
                    return True
            elif item == r:
                return True

    result = []
    for it in inputs:
        if not contained(it):
            result.append(it)
    return result


def merge_recursive(source, destination, none_values=None, overwrite=False):
    if none_values is None:
        none_values = [None]
    for key, value in source.items():
        if isinstance(value, dict):
            # get node or create one
            node = destination.setdefault(key, {})
            merge_recursive(value, node, none_values=none_values, overwrite=overwrite)
        else:
            from requests.models import CaseInsensitiveDict

            if not isinstance(destination, (dict, CaseInsensitiveDict)):
                LOG.warning(
                    "Destination for merging %s=%s is not dict: %s (%s)",
                    key,
                    value,
                    destination,
                    type(destination),
                )
            if overwrite or destination.get(key) in none_values:
                destination[key] = value
    return destination


def merge_dicts(*dicts, **kwargs):
    """Merge all dicts in `*dicts` into a single dict, and return the result. If any of the entries
    in `*dicts` is None, and `default` is specified as keyword argument, then return `default`."""
    result = {}
    for d in dicts:
        if d is None and "default" in kwargs:
            return kwargs["default"]
        if d:
            result.update(d)
    return result


def remove_none_values_from_dict(dict: dict) -> dict:
    return {k: v for (k, v) in dict.items() if v is not None}


def last_index_of(array, value):
    """Return the last index of `value` in the given list, or -1 if it does not exist."""
    result = -1
    for i in reversed(range(len(array))):
        entry = array[i]
        if entry == value or (callable(value) and value(entry)):
            return i
    return result


def is_sub_dict(child_dict: dict, parent_dict: dict) -> bool:
    """Returns whether the first dict is a sub-dict (subset) of the second dict."""
    return all(parent_dict.get(key) == val for key, val in child_dict.items())


def items_equivalent(list1, list2, comparator):
    """Returns whether two lists are equivalent (i.e., same items contained in both lists,
    irrespective of the items' order) with respect to a comparator function."""

    def contained(item):
        for _item in list2:
            if comparator(item, _item):
                return True

    if len(list1) != len(list2):
        return False
    for item in list1:
        if not contained(item):
            return False
    return True


def is_none_or_empty(obj: Union[Optional[str], Optional[list]]) -> bool:
    return (
        obj is None
        or (isinstance(obj, str) and obj.strip() == "")
        or (isinstance(obj, Sized) and len(obj) == 0)
    )


def select_from_typed_dict(typed_dict: type[TypedDict], obj: dict, filter: bool = False) -> dict:
    """
    Select a subset of attributes from a dictionary based on the keys of a given `TypedDict`.
    :param typed_dict: the `TypedDict` blueprint
    :param obj: the object to filter
    :param filter: if True, remove all keys with an empty (e.g., empty string or dictionary) or `None` value
    :return: the resulting dictionary (it returns a copy)
    """
    selection = select_attributes(
        obj, [*typed_dict.__required_keys__, *typed_dict.__optional_keys__]
    )
    if filter:
        selection = {k: v for k, v in selection.items() if v}
    return selection


T = TypeVar("T", bound=dict)


def convert_to_typed_dict(typed_dict: type[T], obj: dict, strict: bool = False) -> T:
    """
    Converts the given object to the given typed dict (by calling the type constructors).
    Limitations:
    - This does not work for ForwardRefs (type refs in quotes).
    - If a type is a Union, the first type is used for the conversion.
    - The conversion fails for types which cannot be instantiated with the constructor.

    :param typed_dict: to convert the given object to
    :param obj: object to convert matching keys to the types defined in the typed dict
    :param strict: True if a TypeError should be raised in case the conversion fails
    :return: obj converted to the typed dict T
    """
    result = cast(T, select_from_typed_dict(typed_dict, obj, filter=True))
    for key, key_type in typed_dict.__annotations__.items():
        if key in result:
            # If it's a Union, or optional, we extract the first type argument
            if get_origin(key_type) in [Union, Optional]:
                key_type = get_args(key_type)[0]
            # Use duck-typing to check if the dict is a typed dict
            if hasattr(key_type, "__required_keys__") and hasattr(key_type, "__optional_keys__"):
                result[key] = convert_to_typed_dict(key_type, result[key])
            else:
                # Otherwise, we call the type's constructor (on a best-effort basis)
                try:
                    result[key] = key_type(result[key])
                except TypeError as e:
                    if strict:
                        raise e
                    else:
                        LOG.debug("Could not convert %s to %s.", key, key_type)
    return result


def dict_multi_values(elements: Union[list, dict]) -> dict[str, list[Any]]:
    """
    Return a dictionary with the original keys from the list of dictionary and the
    values are the list of values of the original dictionary.
    """
    result_dict = {}
    if isinstance(elements, dict):
        for key, value in elements.items():
            if isinstance(value, list):
                result_dict[key] = value
            else:
                result_dict[key] = [value]
    elif isinstance(elements, list):
        if isinstance(elements[0], list):
            for key, value in elements:
                if key in result_dict:
                    result_dict[key].append(value)
                else:
                    result_dict[key] = [value]
        else:
            result_dict[elements[0]] = elements[1:]
    return result_dict


ItemType = TypeVar("ItemType")


def split_list_by(
    lst: Iterable[ItemType], predicate: Callable[[ItemType], bool]
) -> tuple[list[ItemType], list[ItemType]]:
    truthy, falsy = [], []

    for item in lst:
        if predicate(item):
            truthy.append(item)
        else:
            falsy.append(item)

    return truthy, falsy


def is_comma_delimited_list(string: str, item_regex: Optional[str] = None) -> bool:
    """
    Checks if the given string is a comma-delimited list of items.
    The optional `item_regex` parameter specifies the regex pattern for each item in the list.
    """
    item_regex = item_regex or DEFAULT_REGEX_LIST_ITEM

    pattern = re.compile(rf"^\s*({item_regex})(\s*,\s*{item_regex})*\s*$")
    if pattern.match(string) is None:
        return False
    return True


_E = TypeVar("_E")


def optional_list(condition: bool, items: Iterable[_E]) -> list[_E]:
    """
    Given an iterable, either create a list out of the entire iterable (if `condition` is `True`), or return the empty list.
    >>> print(optional_list(True, [1, 2, 3]))
    [1, 2, 3]
    >>> print(optional_list(False, [1, 2, 3]))
    []
    """
    return list(filter(lambda _: condition, items))

1	"""
2	This package provides custom collection types, as well as tools to analyze
3	and manipulate python collection (dicts, list, sets).
4	"""
5
6	import logging	1✔
7	import re	1✔
8	from collections.abc import Iterable, Iterator, Mapping, Sized	1✔
9	from typing import (	1✔
10	Any,
11	Callable,
12	Optional,
13	TypedDict,
14	TypeVar,
15	Union,
16	cast,
17	get_args,
18	get_origin,
19	)
20
21	import cachetools	1✔
22
23	LOG = logging.getLogger(__name__)	1✔
24
25	# default regex to match an item in a comma-separated list string
26	DEFAULT_REGEX_LIST_ITEM = r"[\w-]+"	1✔
27
28
29	class AccessTrackingDict(dict):	1✔
30	"""
31	Simple utility class that can be used to track (write) accesses to a dict's attributes.
32	Note: could also be written as a proxy, to preserve the identity of "wrapped" - for now, it
33	simply duplicates the entries of "wrapped" in the constructor, for simplicity.
34	"""
35
36	def __init__(self, wrapped, callback: Callable[[dict, str, list, dict], Any] = None):	1✔
UNCOV 37	super().__init__(wrapped)	×
UNCOV 38	self.callback = callback	×
39
40	def __setitem__(self, key, value):	1✔
UNCOV 41	self.callback and self.callback(self, "__setitem__", [key, value], {})	×
UNCOV 42	return super().__setitem__(key, value)	×
43
44
45	class DelSafeDict(dict):	1✔
46	"""Useful when applying jsonpatch. Use it as follows:
47
48	obj.__dict__ = DelSafeDict(obj.__dict__)
49	apply_patch(obj.__dict__, patch)
50	"""
51
52	def __delitem__(self, key, args, *kwargs):	1✔
53	self[key] = None	1✔
54
55
56	class ImmutableList(tuple):	1✔
57	"""
58	Wrapper class to create an immutable view of a given list or sequence.
59	Note: Currently, this is simply a wrapper around `tuple` - could be replaced with
60	custom implementations over time, if needed.
61	"""
62
63
64	class HashableList(ImmutableList):	1✔
65	"""Hashable, immutable list wrapper that can be used with dicts or hash sets."""
66
67	def __hash__(self):	1✔
68	return sum(hash(i) for i in self)	1✔
69
70
71	class ImmutableDict(Mapping):	1✔
72	"""Wrapper class to create an immutable view of a given list or sequence."""
73
74	def __init__(self, seq=None, **kwargs):	1✔
75	self._dict = dict(seq, **kwargs)	1✔
76
77	def __len__(self) -> int:	1✔
78	return self._dict.__len__()	1✔
79
80	def __iter__(self) -> Iterator:	1✔
81	return self._dict.__iter__()	1✔
82
83	def __getitem__(self, key):	1✔
84	return self._dict.__getitem__(key)	1✔
85
86	def __eq__(self, other):	1✔
87	return self._dict.__eq__(other._dict if isinstance(other, ImmutableDict) else other)	1✔
88
89	def __str__(self):	1✔
UNCOV 90	return self._dict.__str__()	×
91
92
93	class HashableJsonDict(ImmutableDict):	1✔
94	"""
95	Simple dict wrapper that can be used with dicts or hash sets. Note: the assumption is that the dict
96	can be JSON-encoded (i.e., must be acyclic and contain only lists/dicts and simple types)
97	"""
98
99	def __hash__(self):	1✔
100	from localstack.utils.json import canonical_json	1✔
101
102	return hash(canonical_json(self._dict))	1✔
103
104
105	_ListType = TypeVar("_ListType")	1✔
106
107
108	class PaginatedList(list[_ListType]):	1✔
109	"""List which can be paginated and filtered. For usage in AWS APIs with paginated responses"""
110
111	DEFAULT_PAGE_SIZE = 50	1✔
112
113	def get_page(	1✔
114	self,
115	token_generator: Callable[[_ListType], str],
116	next_token: str = None,
117	page_size: int = None,
118	filter_function: Callable[[_ListType], bool] = None,
119	) -> tuple[list[_ListType], Optional[str]]:
120	if filter_function is not None:	1✔
121	result_list = list(filter(filter_function, self))	1✔
122	else:
123	result_list = self	1✔
124
125	if page_size is None:	1✔
126	page_size = self.DEFAULT_PAGE_SIZE	1✔
127
128	# returns all or remaining elements in final page.
129	if len(result_list) <= page_size and next_token is None:	1✔
130	return result_list, None	1✔
131
132	start_idx = 0	1✔
133
134	try:	1✔
135	start_item = next(item for item in result_list if token_generator(item) == next_token)	1✔
136	start_idx = result_list.index(start_item)	1✔
137	except StopIteration:	1✔
138	pass	1✔
139
140	if start_idx + page_size < len(result_list):	1✔
141	next_token = token_generator(result_list[start_idx + page_size])	1✔
142	else:
143	next_token = None	1✔
144
145	return result_list[start_idx : start_idx + page_size], next_token	1✔
146
147
148	class CustomExpiryTTLCache(cachetools.TTLCache):	1✔
149	"""TTLCache that allows to set custom expiry times for individual keys."""
150
151	def set_expiry(self, key: Any, ttl: Union[float, int]) -> float:	1✔
152	"""Set the expiry of the given key in a TTLCache to (<current_time> + <ttl>)"""
153	with self.timer as time:	1✔
154	# note: need to access the internal dunder API here
155	self._TTLCache__getlink(key).expires = expiry = time + ttl	1✔
156	return expiry	1✔
157
158
159	def get_safe(dictionary, path, default_value=None):	1✔
160	"""
161	Performs a safe navigation on a Dictionary object and
162	returns the result or default value (if specified).
163	The function follows a common AWS path resolution pattern "$.a.b.c".
164
165	:type dictionary: dict
166	:param dictionary: Dict to perform safe navigation.
167
168	:type path: list\|str
169	:param path: List or dot-separated string containing the path of an attribute,
170	starting from the root node "$".
171
172	:type default_value: any
173	:param default_value: Default value to return in case resolved value is None.
174
175	:rtype: any
176	:return: Resolved value or default_value.
177	"""
178	if not isinstance(dictionary, dict) or len(dictionary) == 0:	1✔
179	return default_value	1✔
180
181	attribute_path = path if isinstance(path, list) else path.split(".")	1✔
182	if len(attribute_path) == 0 or attribute_path[0] != "$":	1✔
UNCOV 183	raise AttributeError('Safe navigation must begin with a root node "$"')	×
184
185	current_value = dictionary	1✔
186	for path_node in attribute_path:	1✔
187	if path_node == "$":	1✔
188	continue	1✔
189
190	if re.compile("^\\d+$").search(str(path_node)):	1✔
191	path_node = int(path_node)	1✔
192
193	if isinstance(current_value, dict) and path_node in current_value:	1✔
194	current_value = current_value[path_node]	1✔
195	elif isinstance(current_value, list) and path_node < len(current_value):	1✔
196	current_value = current_value[path_node]	1✔
197	else:
198	current_value = None	1✔
199
200	return current_value or default_value	1✔
201
202
203	def set_safe_mutable(dictionary, path, value):	1✔
204	"""
205	Mutates original dict and sets the specified value under provided path.
206
207	:type dictionary: dict
208	:param dictionary: Dict to mutate.
209
210	:type path: list\|str
211	:param path: List or dot-separated string containing the path of an attribute,
212	starting from the root node "$".
213
214	:type value: any
215	:param value: Value to set under specified path.
216
217	:rtype: dict
218	:return: Returns mutated dictionary.
219	"""
220	if not isinstance(dictionary, dict):	1✔
UNCOV 221	raise AttributeError('"dictionary" must be of type "dict"')	×
222
223	attribute_path = path if isinstance(path, list) else path.split(".")	1✔
224	attribute_path_len = len(attribute_path)	1✔
225
226	if attribute_path_len == 0 or attribute_path[0] != "$":	1✔
UNCOV 227	raise AttributeError('Dict navigation must begin with a root node "$"')	×
228
229	current_pointer = dictionary	1✔
230	for i in range(attribute_path_len):	1✔
231	path_node = attribute_path[i]	1✔
232
233	if path_node == "$":	1✔
234	continue	1✔
235
236	if i < attribute_path_len - 1:	1✔
237	if path_node not in current_pointer:	1✔
238	current_pointer[path_node] = {}	1✔
239	if not isinstance(current_pointer, dict):	1✔
UNCOV 240	raise RuntimeError(	×
241	'Error while deeply setting a dict value. Supplied path is not of type "dict"'
242	)
243	else:
244	current_pointer[path_node] = value	1✔
245
246	current_pointer = current_pointer[path_node]	1✔
247
248	return dictionary	1✔
249
250
251	def pick_attributes(dictionary, paths):	1✔
252	"""
253	Picks selected attributes a returns them as a new dictionary.
254	This function works as a whitelist of attributes to keep in a new dictionary.
255
256	:type dictionary: dict
257	:param dictionary: Dict to pick attributes from.
258
259	:type paths: list of (list or str)
260	:param paths: List of lists or strings with dot-separated paths, starting from the root node "$".
261
262	:rtype: dict
263	:return: Returns whitelisted dictionary.
264	"""
265	new_dictionary = {}	1✔
266
267	for path in paths:	1✔
268	value = get_safe(dictionary, path)	1✔
269
270	if value is not None:	1✔
271	set_safe_mutable(new_dictionary, path, value)	1✔
272
273	return new_dictionary	1✔
274
275
276	def select_attributes(obj: dict, attributes: list[str]) -> dict:	1✔
277	"""Select a subset of attributes from the given dict (returns a copy)"""
278	attributes = attributes if is_list_or_tuple(attributes) else [attributes]	1✔
279	return {k: v for k, v in obj.items() if k in attributes}	1✔
280
281
282	def remove_attributes(obj: dict, attributes: list[str], recursive: bool = False) -> dict:	1✔
283	"""Remove a set of attributes from the given dict (in-place)"""
284	from localstack.utils.objects import recurse_object	1✔
285
286	if recursive:	1✔
287
UNCOV 288	def _remove(o, **kwargs):	×
UNCOV 289	if isinstance(o, dict):	×
UNCOV 290	remove_attributes(o, attributes)	×
UNCOV 291	return o	×
292
UNCOV 293	return recurse_object(obj, _remove)	×
294
295	attributes = ensure_list(attributes)	1✔
296	for attr in attributes:	1✔
297	obj.pop(attr, None)	1✔
298	return obj	1✔
299
300
301	def rename_attributes(	1✔
302	obj: dict, old_to_new_attributes: dict[str, str], in_place: bool = False
303	) -> dict:
304	"""Rename a set of attributes in the given dict object. Second parameter is a dict that maps old to
305	new attribute names. Default is to return a copy, but can also pass in_place=True."""
UNCOV 306	if not in_place:	×
UNCOV 307	obj = dict(obj)	×
UNCOV 308	for old_name, new_name in old_to_new_attributes.items():	×
UNCOV 309	if old_name in obj:	×
UNCOV 310	obj[new_name] = obj.pop(old_name)	×
UNCOV 311	return obj	×
312
313
314	def is_list_or_tuple(obj) -> bool:	1✔
315	return isinstance(obj, (list, tuple))	1✔
316
317
318	def ensure_list(obj: Any, wrap_none=False) -> Optional[list]:	1✔
319	"""Wrap the given object in a list, or return the object itself if it already is a list."""
320	if obj is None and not wrap_none:	1✔
321	return obj	1✔
322	return obj if isinstance(obj, list) else [obj]	1✔
323
324
325	def to_unique_items_list(inputs, comparator=None):	1✔
326	"""Return a list of unique items from the given input iterable.
327	The comparator(item1, item2) returns True/False or an int for comparison."""
328
329	def contained(item):	1✔
330	for r in result:	1✔
331	if comparator:	1✔
332	cmp_res = comparator(item, r)	1✔
333	if cmp_res is True or str(cmp_res) == "0":	1✔
334	return True	1✔
335	elif item == r:	1✔
336	return True	1✔
337
338	result = []	1✔
339	for it in inputs:	1✔
340	if not contained(it):	1✔
341	result.append(it)	1✔
342	return result	1✔
343
344
345	def merge_recursive(source, destination, none_values=None, overwrite=False):	1✔
346	if none_values is None:	1✔
347	none_values = [None]	1✔
348	for key, value in source.items():	1✔
349	if isinstance(value, dict):	1✔
350	# get node or create one
351	node = destination.setdefault(key, {})	1✔
352	merge_recursive(value, node, none_values=none_values, overwrite=overwrite)	1✔
353	else:
354	from requests.models import CaseInsensitiveDict	1✔
355
356	if not isinstance(destination, (dict, CaseInsensitiveDict)):	1✔
UNCOV 357	LOG.warning(	×
358	"Destination for merging %s=%s is not dict: %s (%s)",
359	key,
360	value,
361	destination,
362	type(destination),
363	)
364	if overwrite or destination.get(key) in none_values:	1✔
365	destination[key] = value	1✔
366	return destination	1✔
367
368
369	def merge_dicts(dicts, *kwargs):	1✔
370	"""Merge all dicts in `*dicts` into a single dict, and return the result. If any of the entries
371	in `*dicts` is None, and `default` is specified as keyword argument, then return `default`."""
UNCOV 372	result = {}	×
UNCOV 373	for d in dicts:	×
UNCOV 374	if d is None and "default" in kwargs:	×
UNCOV 375	return kwargs["default"]	×
UNCOV 376	if d:	×
UNCOV 377	result.update(d)	×
UNCOV 378	return result	×
379
380
381	def remove_none_values_from_dict(dict: dict) -> dict:	1✔
382	return {k: v for (k, v) in dict.items() if v is not None}	1✔
383
384
385	def last_index_of(array, value):	1✔
386	"""Return the last index of `value` in the given list, or -1 if it does not exist."""
UNCOV 387	result = -1	×
UNCOV 388	for i in reversed(range(len(array))):	×
UNCOV 389	entry = array[i]	×
UNCOV 390	if entry == value or (callable(value) and value(entry)):	×
UNCOV 391	return i	×
UNCOV 392	return result	×
393
394
395	def is_sub_dict(child_dict: dict, parent_dict: dict) -> bool:	1✔
396	"""Returns whether the first dict is a sub-dict (subset) of the second dict."""
397	return all(parent_dict.get(key) == val for key, val in child_dict.items())	×
398
399
400	def items_equivalent(list1, list2, comparator):	1✔
401	"""Returns whether two lists are equivalent (i.e., same items contained in both lists,
402	irrespective of the items' order) with respect to a comparator function."""
403
404	def contained(item):	×
UNCOV 405	for _item in list2:	×
UNCOV 406	if comparator(item, _item):	×
UNCOV 407	return True	×
408
UNCOV 409	if len(list1) != len(list2):	×
UNCOV 410	return False	×
411	for item in list1:	×
412	if not contained(item):	×
413	return False	×
414	return True	×
415
416
417	def is_none_or_empty(obj: Union[Optional[str], Optional[list]]) -> bool:	1✔
418	return (	1✔
419	obj is None
420	or (isinstance(obj, str) and obj.strip() == "")
421	or (isinstance(obj, Sized) and len(obj) == 0)
422	)
423
424
425	def select_from_typed_dict(typed_dict: type[TypedDict], obj: dict, filter: bool = False) -> dict:	1✔
426	"""
427	Select a subset of attributes from a dictionary based on the keys of a given `TypedDict`.
428	:param typed_dict: the `TypedDict` blueprint
429	:param obj: the object to filter
430	:param filter: if True, remove all keys with an empty (e.g., empty string or dictionary) or `None` value
431	:return: the resulting dictionary (it returns a copy)
432	"""
433	selection = select_attributes(	1✔
434	obj, [typed_dict.__required_keys__, typed_dict.__optional_keys__]
435	)
436	if filter:	1✔
437	selection = {k: v for k, v in selection.items() if v}	1✔
438	return selection	1✔
439
440
441	T = TypeVar("T", bound=dict)	1✔
442
443
444	def convert_to_typed_dict(typed_dict: type[T], obj: dict, strict: bool = False) -> T:	1✔
445	"""
446	Converts the given object to the given typed dict (by calling the type constructors).
447	Limitations:
448	- This does not work for ForwardRefs (type refs in quotes).
449	- If a type is a Union, the first type is used for the conversion.
450	- The conversion fails for types which cannot be instantiated with the constructor.
451
452	:param typed_dict: to convert the given object to
453	:param obj: object to convert matching keys to the types defined in the typed dict
454	:param strict: True if a TypeError should be raised in case the conversion fails
455	:return: obj converted to the typed dict T
456	"""
457	result = cast(T, select_from_typed_dict(typed_dict, obj, filter=True))	1✔
458	for key, key_type in typed_dict.__annotations__.items():	1✔
459	if key in result:	1✔
460	# If it's a Union, or optional, we extract the first type argument
461	if get_origin(key_type) in [Union, Optional]:	1✔
462	key_type = get_args(key_type)[0]	1✔
463	# Use duck-typing to check if the dict is a typed dict
464	if hasattr(key_type, "__required_keys__") and hasattr(key_type, "__optional_keys__"):	1✔
465	result[key] = convert_to_typed_dict(key_type, result[key])	1✔
466	else:
467	# Otherwise, we call the type's constructor (on a best-effort basis)
468	try:	1✔
469	result[key] = key_type(result[key])	1✔
470	except TypeError as e:	1✔
471	if strict:	1✔
472	raise e	1✔
473	else:
474	LOG.debug("Could not convert %s to %s.", key, key_type)	1✔
475	return result	1✔
476
477
478	def dict_multi_values(elements: Union[list, dict]) -> dict[str, list[Any]]:	1✔
479	"""
480	Return a dictionary with the original keys from the list of dictionary and the
481	values are the list of values of the original dictionary.
482	"""
483	result_dict = {}	1✔
484	if isinstance(elements, dict):	1✔
485	for key, value in elements.items():	1✔
486	if isinstance(value, list):	1✔
487	result_dict[key] = value	1✔
488	else:
489	result_dict[key] = [value]	1✔
490	elif isinstance(elements, list):	1✔
491	if isinstance(elements[0], list):	1✔
492	for key, value in elements:	1✔
493	if key in result_dict:	1✔
494	result_dict[key].append(value)	1✔
495	else:
496	result_dict[key] = [value]	1✔
497	else:
498	result_dict[elements[0]] = elements[1:]	1✔
499	return result_dict	1✔
500
501
502	ItemType = TypeVar("ItemType")	1✔
503
504
505	def split_list_by(	1✔
506	lst: Iterable[ItemType], predicate: Callable[[ItemType], bool]
507	) -> tuple[list[ItemType], list[ItemType]]:
UNCOV 508	truthy, falsy = [], []	×
509
UNCOV 510	for item in lst:	×
UNCOV 511	if predicate(item):	×
UNCOV 512	truthy.append(item)	×
513	else:
UNCOV 514	falsy.append(item)	×
515
UNCOV 516	return truthy, falsy	×
517
518
519	def is_comma_delimited_list(string: str, item_regex: Optional[str] = None) -> bool:	1✔
520	"""
521	Checks if the given string is a comma-delimited list of items.
522	The optional `item_regex` parameter specifies the regex pattern for each item in the list.
523	"""
524	item_regex = item_regex or DEFAULT_REGEX_LIST_ITEM	1✔
525
526	pattern = re.compile(rf"^\s({item_regex})(\s,\s{item_regex})\s*$")	1✔
527	if pattern.match(string) is None:	1✔
528	return False	1✔
529	return True	1✔
530
531
532	_E = TypeVar("_E")	1✔
533
534
535	def optional_list(condition: bool, items: Iterable[_E]) -> list[_E]:	1✔
536	"""
537	Given an iterable, either create a list out of the entire iterable (if `condition` is `True`), or return the empty list.
538	>>> print(optional_list(True, [1, 2, 3]))
539	[1, 2, 3]
540	>>> print(optional_list(False, [1, 2, 3]))
541	[]
542	"""
543	return list(filter(lambda _: condition, items))	1✔

localstack / localstack / 16820655284

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous