18917981582

Committed 29 Oct 2025 06:16PM UTC coverage: 79.819% (-0.2%) from 80.004%

Build # 18917981582

Build Type

Pull #22837

github

Committed by

web-flow

Commit Message

Merge 05341325b into 1a7da5c5e

Pull Request Pull Request #22837: Updated Treesitter dependencies

Run Details

77044 of 96523 relevant lines covered (79.82%)

3.05 hits per line

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

78.52

/src/python/pants/backend/helm/utils/yaml.py

# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

from abc import ABCMeta
from collections import defaultdict
from collections.abc import Callable, Iterable, Iterator, Mapping
from dataclasses import dataclass
from pathlib import PurePath
from typing import Any, Generic, TypeVar

from pants.engine.collection import Collection
from pants.util.frozendict import FrozenDict


@dataclass(unsafe_hash=True)
class YamlPath:
    """Simple implementation of YAML paths using `/` syntax and being the single slash the path to
    the root."""

    _elements: tuple[str, ...]
    _absolute: bool

    def __init__(self, elements: Iterable[str], *, absolute: bool) -> None:
        object.__setattr__(self, "_elements", tuple(elements))
        object.__setattr__(self, "_absolute", absolute)

        if len(self._elements) == 0 and not self._absolute:
            raise ValueError("Relative YAML paths with no elements are not allowed.")

    @classmethod
    def parse(cls, path: str) -> YamlPath:
        """Parses a YAML path."""

        is_absolute = path.startswith("/")
        return cls([elem for elem in path.split("/") if elem], absolute=is_absolute)

    @classmethod
    def root(cls) -> YamlPath:
        """Returns a YamlPath that represents the root element."""

        return cls([], absolute=True)

    @classmethod
    def index(cls, idx: int) -> YamlPath:
        """Returns a relative YamlPath for the index value provided."""

        return cls([str(idx)], absolute=False)

    @property
    def parent(self) -> YamlPath | None:
        """Returns the path to the parent element unless this path is already the root."""

        if not self.is_root:
            return YamlPath(self._elements[:-1], absolute=self._absolute)
        return None

    @property
    def current(self) -> str:
        """Returns the name of the current element referenced by this path.

        The root element will return the empty string.
        """

        if self.is_root:
            return ""
        return self._elements[len(self._elements) - 1]

    @property
    def is_absolute(self) -> bool:
        """Returns `True` if this is an absolute path."""

        return self._absolute

    @property
    def is_root(self) -> bool:
        """Returns `True` if this path represents the root element."""

        return len(self._elements) == 0 and self._absolute

    @property
    def is_index(self) -> bool:
        """Returns `True` if this path is referencing an indexed item inside an array."""

        try:
            int(self.current)
            return True
        except ValueError:
            return False

    def to_relative(self) -> YamlPath:
        """Transforms this YamlPath instance into a relative path."""

        if not self._absolute:
            return self
        return YamlPath(self._elements, absolute=False)

    def __truediv__(self, other: str | int | YamlPath) -> YamlPath:
        if isinstance(other, str):
            other_path = YamlPath.parse(other)
        elif isinstance(other, int):
            other_path = YamlPath.index(other)
        else:
            other_path = other

        if other_path._absolute:
            raise ValueError("Can not append an absolute path to another path.")

        return YamlPath(self._elements + other_path._elements, absolute=self._absolute)

    def __iter__(self):
        return iter(self._elements)

    def __str__(self) -> str:
        path = "/".join(self._elements)
        if self._absolute:
            path = f"/{path}"
        return path


@dataclass(frozen=True)
class YamlElement(metaclass=ABCMeta):
    """Abstract base class for elements read from YAML files.

    `element_path` represents the location inside the YAML file where this element is.
    """

    element_path: YamlPath


T = TypeVar("T")
R = TypeVar("R")


class MutableYamlIndex(Generic[T]):
    """Represents a mutable collection of items that is indexed by the following keys:

    - the relative path of the YAML file
    - the document index inside the YAML file
    - the YAML path of the item
    """

    _data: dict[PurePath, dict[int, dict[YamlPath, T]]]

    def __init__(self) -> None:
        self._data = defaultdict(dict)

    def insert(
        self, *, file_path: PurePath, yaml_path: YamlPath, item: T, document_index: int = 0
    ) -> None:
        """Inserts an item at the given position in the index."""

        doc_index = self._data[file_path].get(document_index, {})
        if not doc_index:
            self._data[file_path][document_index] = doc_index

        doc_index[yaml_path] = item

    def frozen(self) -> FrozenYamlIndex[T]:
        """Transforms this collection into a frozen (immutable) one."""

        return FrozenYamlIndex.create(self)


@dataclass(frozen=True)
class _YamlDocumentIndexNode(Generic[T]):
    """Helper node item for the `FrozenYamlIndex` type."""

    paths: FrozenDict[YamlPath, T]

    @classmethod
    def empty(cls: type[_YamlDocumentIndexNode[T]]) -> _YamlDocumentIndexNode[T]:
        return cls(paths=FrozenDict())

    def to_json_dict(self) -> dict[str, dict[str, str]]:
        items_dict: dict[str, str] = {}
        for path, item in self.paths.items():
            items_dict[str(path)] = str(item)
        return {"paths": items_dict}


@dataclass(frozen=True)
class FrozenYamlIndex(Generic[T]):
    """Represents a frozen collection of items that is indexed by the following keys:

    - the relative path of the YAML file
    - the document index inside the YAML file
    - the YAML path of the item
    """

    _data: FrozenDict[PurePath, Collection[_YamlDocumentIndexNode[T]]]

    @classmethod
    def create(cls, other: MutableYamlIndex[T]) -> FrozenYamlIndex[T]:
        data: dict[PurePath, Collection[_YamlDocumentIndexNode[T]]] = {}
        for file_path, doc_index in other._data.items():
            max_index = max(doc_index.keys())
            doc_list: list[_YamlDocumentIndexNode[T]] = [_YamlDocumentIndexNode.empty()] * (
                max_index + 1
            )

            for idx, item_map in doc_index.items():
                doc_list[idx] = _YamlDocumentIndexNode(paths=FrozenDict(item_map))

            data[file_path] = Collection(doc_list)
        return FrozenYamlIndex(_data=FrozenDict(data))

    @classmethod
    def empty(cls: type[FrozenYamlIndex[T]]) -> FrozenYamlIndex[T]:
        return FrozenYamlIndex[T](_data=FrozenDict())

    def transform_values(self, func: Callable[[T], R | None]) -> FrozenYamlIndex[R]:
        """Transforms the values of the given indexed collection into those that are returned from
        the received function.

        The items that map to `None` in the given function are not included in the result.

        This is a combination of the `map` and `filter` higher-order functions into one so
        both operations are performed in a single pass.
        """

        mutable_index: MutableYamlIndex[R] = MutableYamlIndex()
        for file_path, doc_index, yaml_path, item in self:
            new_item = func(item)
            if new_item is not None:
                mutable_index.insert(
                    file_path=file_path,
                    document_index=doc_index,
                    yaml_path=yaml_path,
                    item=new_item,
                )
        return mutable_index.frozen()

    def values(self) -> Iterator[T]:
        """Returns an iterator over the values of this index."""
        for _, _, _, item in self:
            yield item

    def to_json_dict(self) -> dict[str, Any]:
        """Transforms this collection into a JSON-like dictionary that can be dumped later."""

        result = {}
        for file_path, documents in self._data.items():
            result[str(file_path)] = [doc_idx.to_json_dict() for doc_idx in documents]
        return result

    def __iter__(self):
        for file_path, doc_indexes in self._data.items():
            for idx, doc_index in enumerate(doc_indexes):
                for yaml_path, item in doc_index.paths.items():
                    yield file_path, idx, yaml_path, item


def _to_snake_case(str: str) -> str:
    """Translates a camel-case or kebab-case identifier into a snake-case one."""

    base_string = str.replace("-", "_")

    result = ""
    idx = 0
    for c in base_string:
        char_to_add = c
        if char_to_add.isupper():
            char_to_add = c.lower()
            if idx > 0:
                result += "_"
        result += char_to_add
        idx += 1

    return result


def snake_case_attr_dict(d: Mapping[str, Any]) -> dict[str, Any]:
    """Transforms all keys in the given mapping to be snake-case."""
    return {_to_snake_case(name): value for name, value in d.items()}

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	8✔
5
6	from abc import ABCMeta	8✔
7	from collections import defaultdict	8✔
8	from collections.abc import Callable, Iterable, Iterator, Mapping	8✔
9	from dataclasses import dataclass	8✔
10	from pathlib import PurePath	8✔
11	from typing import Any, Generic, TypeVar	8✔
12
13	from pants.engine.collection import Collection	8✔
14	from pants.util.frozendict import FrozenDict	8✔
15
16
17	@dataclass(unsafe_hash=True)	8✔
18	class YamlPath:	8✔
19	"""Simple implementation of YAML paths using `/` syntax and being the single slash the path to
20	the root."""
21
22	_elements: tuple[str, ...]	8✔
23	_absolute: bool	8✔
24
25	def __init__(self, elements: Iterable[str], *, absolute: bool) -> None:	8✔
26	object.__setattr__(self, "_elements", tuple(elements))	3✔
27	object.__setattr__(self, "_absolute", absolute)	3✔
28
29	if len(self._elements) == 0 and not self._absolute:	3✔
30	raise ValueError("Relative YAML paths with no elements are not allowed.")	×
31
32	@classmethod	8✔
33	def parse(cls, path: str) -> YamlPath:	8✔
34	"""Parses a YAML path."""
35
36	is_absolute = path.startswith("/")	3✔
37	return cls([elem for elem in path.split("/") if elem], absolute=is_absolute)	3✔
38
39	@classmethod	8✔
40	def root(cls) -> YamlPath:	8✔
41	"""Returns a YamlPath that represents the root element."""
42
43	return cls([], absolute=True)	1✔
44
45	@classmethod	8✔
46	def index(cls, idx: int) -> YamlPath:	8✔
47	"""Returns a relative YamlPath for the index value provided."""
48
49	return cls([str(idx)], absolute=False)	1✔
50
51	@property	8✔
52	def parent(self) -> YamlPath \| None:	8✔
53	"""Returns the path to the parent element unless this path is already the root."""
54
55	if not self.is_root:	1✔
56	return YamlPath(self._elements[:-1], absolute=self._absolute)	1✔
57	return None	1✔
58
59	@property	8✔
60	def current(self) -> str:	8✔
61	"""Returns the name of the current element referenced by this path.
62
63	The root element will return the empty string.
64	"""
65
66	if self.is_root:	1✔
67	return ""	1✔
68	return self._elements[len(self._elements) - 1]	1✔
69
70	@property	8✔
71	def is_absolute(self) -> bool:	8✔
72	"""Returns `True` if this is an absolute path."""
73
74	return self._absolute	1✔
75
76	@property	8✔
77	def is_root(self) -> bool:	8✔
78	"""Returns `True` if this path represents the root element."""
79
80	return len(self._elements) == 0 and self._absolute	1✔
81
82	@property	8✔
83	def is_index(self) -> bool:	8✔
84	"""Returns `True` if this path is referencing an indexed item inside an array."""
85
86	try:	1✔
87	int(self.current)	1✔
88	return True	1✔
89	except ValueError:	1✔
90	return False	1✔
91
92	def to_relative(self) -> YamlPath:	8✔
93	"""Transforms this YamlPath instance into a relative path."""
94
95	if not self._absolute:	×
96	return self	×
97	return YamlPath(self._elements, absolute=False)	×
98
99	def __truediv__(self, other: str \| int \| YamlPath) -> YamlPath:	8✔
100	if isinstance(other, str):	1✔
101	other_path = YamlPath.parse(other)	1✔
102	elif isinstance(other, int):	1✔
103	other_path = YamlPath.index(other)	×
104	else:
105	other_path = other	1✔
106
107	if other_path._absolute:	1✔
108	raise ValueError("Can not append an absolute path to another path.")	×
109
110	return YamlPath(self._elements + other_path._elements, absolute=self._absolute)	1✔
111
112	def __iter__(self):	8✔
113	return iter(self._elements)	×
114
115	def __str__(self) -> str:	8✔
116	path = "/".join(self._elements)	×
117	if self._absolute:	×
118	path = f"/{path}"	×
119	return path	×
120
121
122	@dataclass(frozen=True)	8✔
123	class YamlElement(metaclass=ABCMeta):	8✔
124	"""Abstract base class for elements read from YAML files.
125
126	`element_path` represents the location inside the YAML file where this element is.
127	"""
128
129	element_path: YamlPath	8✔
130
131
132	T = TypeVar("T")	8✔
133	R = TypeVar("R")	8✔
134
135
136	class MutableYamlIndex(Generic[T]):	8✔
137	"""Represents a mutable collection of items that is indexed by the following keys:
138
139	- the relative path of the YAML file
140	- the document index inside the YAML file
141	- the YAML path of the item
142	"""
143
144	_data: dict[PurePath, dict[int, dict[YamlPath, T]]]	8✔
145
146	def __init__(self) -> None:	8✔
147	self._data = defaultdict(dict)	1✔
148
149	def insert(	8✔
150	self, *, file_path: PurePath, yaml_path: YamlPath, item: T, document_index: int = 0
151	) -> None:
152	"""Inserts an item at the given position in the index."""
153
154	doc_index = self._data[file_path].get(document_index, {})	1✔
155	if not doc_index:	1✔
156	self._data[file_path][document_index] = doc_index	1✔
157
158	doc_index[yaml_path] = item	1✔
159
160	def frozen(self) -> FrozenYamlIndex[T]:	8✔
161	"""Transforms this collection into a frozen (immutable) one."""
162
163	return FrozenYamlIndex.create(self)	1✔
164
165
166	@dataclass(frozen=True)	8✔
167	class _YamlDocumentIndexNode(Generic[T]):	8✔
168	"""Helper node item for the `FrozenYamlIndex` type."""
169
170	paths: FrozenDict[YamlPath, T]	8✔
171
172	@classmethod	8✔
173	def empty(cls: type[_YamlDocumentIndexNode[T]]) -> _YamlDocumentIndexNode[T]:	8✔
174	return cls(paths=FrozenDict())	1✔
175
176	def to_json_dict(self) -> dict[str, dict[str, str]]:	8✔
177	items_dict: dict[str, str] = {}	×
178	for path, item in self.paths.items():	×
179	items_dict[str(path)] = str(item)	×
180	return {"paths": items_dict}	×
181
182
183	@dataclass(frozen=True)	8✔
184	class FrozenYamlIndex(Generic[T]):	8✔
185	"""Represents a frozen collection of items that is indexed by the following keys:
186
187	- the relative path of the YAML file
188	- the document index inside the YAML file
189	- the YAML path of the item
190	"""
191
192	_data: FrozenDict[PurePath, Collection[_YamlDocumentIndexNode[T]]]	8✔
193
194	@classmethod	8✔
195	def create(cls, other: MutableYamlIndex[T]) -> FrozenYamlIndex[T]:	8✔
196	data: dict[PurePath, Collection[_YamlDocumentIndexNode[T]]] = {}	1✔
197	for file_path, doc_index in other._data.items():	1✔
198	max_index = max(doc_index.keys())	1✔
199	doc_list: list[_YamlDocumentIndexNode[T]] = [_YamlDocumentIndexNode.empty()] * (	1✔
200	max_index + 1
201	)
202
203	for idx, item_map in doc_index.items():	1✔
204	doc_list[idx] = _YamlDocumentIndexNode(paths=FrozenDict(item_map))	1✔
205
206	data[file_path] = Collection(doc_list)	1✔
207	return FrozenYamlIndex(_data=FrozenDict(data))	1✔
208
209	@classmethod	8✔
210	def empty(cls: type[FrozenYamlIndex[T]]) -> FrozenYamlIndex[T]:	8✔
211	return FrozenYamlIndex[T](_data=FrozenDict())	×
212
213	def transform_values(self, func: Callable[[T], R \| None]) -> FrozenYamlIndex[R]:	8✔
214	"""Transforms the values of the given indexed collection into those that are returned from
215	the received function.
216
217	The items that map to `None` in the given function are not included in the result.
218
219	This is a combination of the `map` and `filter` higher-order functions into one so
220	both operations are performed in a single pass.
221	"""
222
223	mutable_index: MutableYamlIndex[R] = MutableYamlIndex()	×
224	for file_path, doc_index, yaml_path, item in self:	×
225	new_item = func(item)	×
226	if new_item is not None:	×
227	mutable_index.insert(	×
228	file_path=file_path,
229	document_index=doc_index,
230	yaml_path=yaml_path,
231	item=new_item,
232	)
233	return mutable_index.frozen()	×
234
235	def values(self) -> Iterator[T]:	8✔
236	"""Returns an iterator over the values of this index."""
237	for _, _, _, item in self:	×
238	yield item	×
239
240	def to_json_dict(self) -> dict[str, Any]:	8✔
241	"""Transforms this collection into a JSON-like dictionary that can be dumped later."""
242
243	result = {}	×
244	for file_path, documents in self._data.items():	×
245	result[str(file_path)] = [doc_idx.to_json_dict() for doc_idx in documents]	×
246	return result	×
247
248	def __iter__(self):	8✔
249	for file_path, doc_indexes in self._data.items():	×
250	for idx, doc_index in enumerate(doc_indexes):	×
251	for yaml_path, item in doc_index.paths.items():	×
252	yield file_path, idx, yaml_path, item	×
253
254
255	def _to_snake_case(str: str) -> str:	8✔
256	"""Translates a camel-case or kebab-case identifier into a snake-case one."""
257
258	base_string = str.replace("-", "_")	1✔
259
260	result = ""	1✔
261	idx = 0	1✔
262	for c in base_string:	1✔
263	char_to_add = c	1✔
264	if char_to_add.isupper():	1✔
265	char_to_add = c.lower()	1✔
266	if idx > 0:	1✔
267	result += "_"	1✔
268	result += char_to_add	1✔
269	idx += 1	1✔
270
271	return result	1✔
272
273
274	def snake_case_attr_dict(d: Mapping[str, Any]) -> dict[str, Any]:	8✔
275	"""Transforms all keys in the given mapping to be snake-case."""
276	return {_to_snake_case(name): value for name, value in d.items()}	1✔

pantsbuild / pants / 18917981582

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