21552830208

Committed 31 Jan 2026 11:40PM UTC coverage: 80.277% (-0.05%) from 80.324%

Build # 21552830208

Build Type

Pull #23062

github

Committed by

web-flow

Commit Message

Merge 808a9786c into 2c4dcf9cf

Pull Request Pull Request #23062: Remove support for Get

Run Details

18 of 25 new or added lines in 4 files covered. (72.0%)

17119 existing lines in 541 files now uncovered.

78278 of 97510 relevant lines covered (80.28%)

3.36 hits per line

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

91.04

/src/python/pants/util/collections.py

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

from __future__ import annotations

import collections
import collections.abc
import gc
import math
from collections.abc import Callable, Iterable, Iterator, MutableMapping
from sys import getsizeof
from typing import Any, TypeVar

from pants.engine.internals import native_engine
from pants.util.strutil import softwrap


def recursively_update(d: MutableMapping, d2: MutableMapping) -> None:
    """dict.update but which merges child dicts (dict2 takes precedence where there's conflict)."""
    for k, v in d2.items():
        if k in d:
            if isinstance(v, dict):
                recursively_update(d[k], v)
                continue
        d[k] = v


def deep_getsizeof(o: Any, ids: set[int]) -> int:
    """Find the memory footprint of the given object.

    To avoid double-counting, `ids` should be a set of object ids which have been visited by
    previous calls to this method.
    """
    if id(o) in ids:
        return 0

    d = deep_getsizeof
    r = getsizeof(o)
    ids.add(id(o))

    return r + sum(d(x, ids) for x in gc.get_referents())


_T = TypeVar("_T")


def assert_single_element(iterable: Iterable[_T]) -> _T:
    """Get the single element of `iterable`, or raise an error.

    :raise: :class:`StopIteration` if there is no element.
    :raise: :class:`ValueError` if there is more than one element.
    """
    it = iter(iterable)
    first_item = next(it)

    try:
        next(it)
    except StopIteration:
        return first_item

    raise ValueError(f"iterable {iterable!r} has more than one element.")


def ensure_list(
    val: Any | Iterable[Any], *, expected_type: type[_T], allow_single_scalar: bool = False
) -> list[_T]:
    """Ensure that every element of an iterable is the expected type and convert the result to a
    list.

    If `allow_single_scalar` is True, a single value T will be wrapped into a `List[T]`.
    """
    if isinstance(val, expected_type):
        if not allow_single_scalar:
            raise ValueError(f"The value {val} must be wrapped in an iterable (e.g. a list).")
        return [val]
    if not isinstance(val, collections.abc.Iterable):
        raise ValueError(
            f"The value {val} (type {type(val)}) was not an iterable of {expected_type}."
        )
    result: list[_T] = []
    for i, x in enumerate(val):
        if not isinstance(x, expected_type):
            raise ValueError(
                softwrap(
                    f"""
                    Not all elements of the iterable have type {expected_type}. Encountered the
                    element {x} of type {type(x)} at index {i}.
                    """
                )
            )
        result.append(x)
    return result


def ensure_str_list(val: str | Iterable[str], *, allow_single_str: bool = False) -> list[str]:
    """Ensure that every element of an iterable is a string and convert the result to a list.

    If `allow_single_str` is True, a single `str` will be wrapped into a `List[str]`.
    """
    return ensure_list(val, expected_type=str, allow_single_scalar=allow_single_str)


def partition_sequentially(
    items: Iterable[_T],
    *,
    key: Callable[[_T], str],
    size_target: int,
    size_max: int | None = None,
) -> Iterator[list[_T]]:
    """Stably partitions the given items into batches of around `size_target` items.

    The "stability" property refers to avoiding adjusting all batches when a single item is added,
    which could happen if the items were trivially windowed using `itertools.islice` and an
    item was added near the front of the list.

    Batches will optionally be capped to `size_max`, but note that this can weaken the stability
    properties of the bucketing, by forcing bucket boundaries to be created where they otherwise
    might not.
    """

    # To stably partition the arguments into ranges of approximately `size_target`, we sort them,
    # and create a new batch sequentially once we encounter an item hash prefixed with a threshold
    # of zeros.
    #
    # The hashes act like a (deterministic) series of rolls of an evenly distributed die. The
    # probability of a hash prefixed with Z zero bits is 1/2^Z, and so to break after N items on
    # average, we look for `Z == log2(N)` zero bits.
    #
    # Breaking on these deterministic boundaries reduces the chance that adding or removing items
    # causes multiple buckets to be recalculated. But when a `size_max` value is set, it's possible
    # for adding items to cause multiple sequential buckets to be affected.
    zero_prefix_threshold = math.log(max(1, size_target), 2)

    batch: list[_T] = []

    def emit_batch() -> list[_T]:
        assert batch
        result = list(batch)
        batch.clear()
        return result

    keyed_items = []
    for item in items:
        keyed_items.append((key(item), item))
    keyed_items.sort()

    for item_key, item in keyed_items:
        batch.append(item)
        prefix_zero_bits = native_engine.hash_prefix_zero_bits(item_key)
        if prefix_zero_bits >= zero_prefix_threshold or (size_max and len(batch) >= size_max):
            yield emit_batch()
    if batch:
        yield emit_batch()

1	# Copyright 2017 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	12✔
5
6	import collections	12✔
7	import collections.abc	12✔
8	import gc	12✔
9	import math	12✔
10	from collections.abc import Callable, Iterable, Iterator, MutableMapping	12✔
11	from sys import getsizeof	12✔
12	from typing import Any, TypeVar	12✔
13
14	from pants.engine.internals import native_engine	12✔
15	from pants.util.strutil import softwrap	12✔
16
17
18	def recursively_update(d: MutableMapping, d2: MutableMapping) -> None:	12✔
19	"""dict.update but which merges child dicts (dict2 takes precedence where there's conflict)."""
20	for k, v in d2.items():	7✔
UNCOV 21	if k in d:	2✔
UNCOV 22	if isinstance(v, dict):	2✔
UNCOV 23	recursively_update(d[k], v)	2✔
UNCOV 24	continue	2✔
UNCOV 25	d[k] = v	2✔
26
27
28	def deep_getsizeof(o: Any, ids: set[int]) -> int:	12✔
29	"""Find the memory footprint of the given object.
30
31	To avoid double-counting, `ids` should be a set of object ids which have been visited by
32	previous calls to this method.
33	"""
34	if id(o) in ids:	×
35	return 0	×
36
37	d = deep_getsizeof	×
38	r = getsizeof(o)	×
39	ids.add(id(o))	×
40
41	return r + sum(d(x, ids) for x in gc.get_referents())	×
42
43
44	_T = TypeVar("_T")	12✔
45
46
47	def assert_single_element(iterable: Iterable[_T]) -> _T:	12✔
48	"""Get the single element of `iterable`, or raise an error.
49
50	:raise: :class:`StopIteration` if there is no element.
51	:raise: :class:`ValueError` if there is more than one element.
52	"""
53	it = iter(iterable)	12✔
54	first_item = next(it)	12✔
55
56	try:	12✔
57	next(it)	12✔
58	except StopIteration:	12✔
59	return first_item	12✔
60
UNCOV 61	raise ValueError(f"iterable {iterable!r} has more than one element.")	1✔
62
63
64	def ensure_list(	12✔
65	val: Any \| Iterable[Any], *, expected_type: type[_T], allow_single_scalar: bool = False
66	) -> list[_T]:
67	"""Ensure that every element of an iterable is the expected type and convert the result to a
68	list.
69
70	If `allow_single_scalar` is True, a single value T will be wrapped into a `List[T]`.
71	"""
72	if isinstance(val, expected_type):	12✔
UNCOV 73	if not allow_single_scalar:	3✔
UNCOV 74	raise ValueError(f"The value {val} must be wrapped in an iterable (e.g. a list).")	2✔
UNCOV 75	return [val]	3✔
76	if not isinstance(val, collections.abc.Iterable):	12✔
UNCOV 77	raise ValueError(	1✔
78	f"The value {val} (type {type(val)}) was not an iterable of {expected_type}."
79	)
80	result: list[_T] = []	12✔
81	for i, x in enumerate(val):	12✔
82	if not isinstance(x, expected_type):	12✔
UNCOV 83	raise ValueError(	2✔
84	softwrap(
85	f"""
86	Not all elements of the iterable have type {expected_type}. Encountered the
87	element {x} of type {type(x)} at index {i}.
88	"""
89	)
90	)
91	result.append(x)	12✔
92	return result	12✔
93
94
95	def ensure_str_list(val: str \| Iterable[str], *, allow_single_str: bool = False) -> list[str]:	12✔
96	"""Ensure that every element of an iterable is a string and convert the result to a list.
97
98	If `allow_single_str` is True, a single `str` will be wrapped into a `List[str]`.
99	"""
UNCOV 100	return ensure_list(val, expected_type=str, allow_single_scalar=allow_single_str)	3✔
101
102
103	def partition_sequentially(	12✔
104	items: Iterable[_T],
105	*,
106	key: Callable[[_T], str],
107	size_target: int,
108	size_max: int \| None = None,
109	) -> Iterator[list[_T]]:
110	"""Stably partitions the given items into batches of around `size_target` items.
111
112	The "stability" property refers to avoiding adjusting all batches when a single item is added,
113	which could happen if the items were trivially windowed using `itertools.islice` and an
114	item was added near the front of the list.
115
116	Batches will optionally be capped to `size_max`, but note that this can weaken the stability
117	properties of the bucketing, by forcing bucket boundaries to be created where they otherwise
118	might not.
119	"""
120
121	# To stably partition the arguments into ranges of approximately `size_target`, we sort them,
122	# and create a new batch sequentially once we encounter an item hash prefixed with a threshold
123	# of zeros.
124	#
125	# The hashes act like a (deterministic) series of rolls of an evenly distributed die. The
126	# probability of a hash prefixed with Z zero bits is 1/2^Z, and so to break after N items on
127	# average, we look for `Z == log2(N)` zero bits.
128	#
129	# Breaking on these deterministic boundaries reduces the chance that adding or removing items
130	# causes multiple buckets to be recalculated. But when a `size_max` value is set, it's possible
131	# for adding items to cause multiple sequential buckets to be affected.
UNCOV 132	zero_prefix_threshold = math.log(max(1, size_target), 2)	3✔
133
UNCOV 134	batch: list[_T] = []	3✔
135
UNCOV 136	def emit_batch() -> list[_T]:	3✔
UNCOV 137	assert batch	3✔
UNCOV 138	result = list(batch)	3✔
UNCOV 139	batch.clear()	3✔
UNCOV 140	return result	3✔
141
UNCOV 142	keyed_items = []	3✔
UNCOV 143	for item in items:	3✔
UNCOV 144	keyed_items.append((key(item), item))	3✔
UNCOV 145	keyed_items.sort()	3✔
146
UNCOV 147	for item_key, item in keyed_items:	3✔
UNCOV 148	batch.append(item)	3✔
UNCOV 149	prefix_zero_bits = native_engine.hash_prefix_zero_bits(item_key)	3✔
UNCOV 150	if prefix_zero_bits >= zero_prefix_threshold or (size_max and len(batch) >= size_max):	3✔
UNCOV 151	yield emit_batch()	3✔
UNCOV 152	if batch:	3✔
UNCOV 153	yield emit_batch()	2✔

pantsbuild / pants / 21552830208

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