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

32.26

/localstack-core/localstack/services/lambda_/event_source_mapping/senders/sender_utils.py

import sys
from collections.abc import Iterable, Iterator
from itertools import islice
from typing import Any


def batched(iterable, n):
    # TODO: replace this method with native version when supporting Python 3.12
    #  https://docs.python.org/3.12/library/itertools.html#itertools.batched
    # batched('ABCDEFG', 3) --> ABC DEF G
    if n < 1:
        raise ValueError("n must be at least one")
    it = iter(iterable)
    while batch := tuple(islice(it, n)):
        yield batch


def batched_by_size(iterable: Iterable[Any], max_bytes) -> Iterator[tuple[Any, ...]]:
    """
    Generate batches from iterable where the total size of each batch in bytes does not exceed `max_bytes`.
    """
    if max_bytes < 1:
        raise ValueError("max_bytes must be at least one")

    it = iter(iterable)
    while True:
        batch = []
        current_size = 0
        try:
            while current_size < max_bytes:
                item = next(it)
                item_size = sys.getsizeof(item)
                if current_size + item_size > max_bytes:
                    # If adding this item exceeds max_bytes, push it back onto the iterator and stop this batch
                    it = iter([item] + list(it))
                    break
                batch.append(item)
                current_size += item_size
        except StopIteration:
            pass

        if not batch:
            break
        yield tuple(batch)

1	import sys	1✔
2	from collections.abc import Iterable, Iterator	1✔
3	from itertools import islice	1✔
4	from typing import Any	1✔
5
6
7	def batched(iterable, n):	1✔
8	# TODO: replace this method with native version when supporting Python 3.12
9	# https://docs.python.org/3.12/library/itertools.html#itertools.batched
10	# batched('ABCDEFG', 3) --> ABC DEF G
11	if n < 1:	1✔
UNCOV 12	raise ValueError("n must be at least one")	×
13	it = iter(iterable)	1✔
14	while batch := tuple(islice(it, n)):	1✔
15	yield batch	1✔
16
17
18	def batched_by_size(iterable: Iterable[Any], max_bytes) -> Iterator[tuple[Any, ...]]:	1✔
19	"""
20	Generate batches from iterable where the total size of each batch in bytes does not exceed `max_bytes`.
21	"""
22	if max_bytes < 1:	×
UNCOV 23	raise ValueError("max_bytes must be at least one")	×
24
25	it = iter(iterable)	×
26	while True:	×
27	batch = []	×
28	current_size = 0	×
29	try:	×
30	while current_size < max_bytes:	×
31	item = next(it)	×
32	item_size = sys.getsizeof(item)	×
UNCOV 33	if current_size + item_size > max_bytes:	×
34	# If adding this item exceeds max_bytes, push it back onto the iterator and stop this batch
35	it = iter([item] + list(it))	×
36	break	×
37	batch.append(item)	×
38	current_size += item_size	×
39	except StopIteration:	×
UNCOV 40	pass	×
41
42	if not batch:	×
43	break	×
UNCOV 44	yield tuple(batch)	×

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