GreptimeTeam/demo-scene

Implement itertools.batched with a function to drop requirement of Python 3.12+ in InfluxDB demo

tisonkun opened this issue · 2 comments

# TODO implement itertools.batched with a functionn to drop requirement of Python 3.12+
for batch_lines in itertools.batched(lines, 1000):
write_api.write(bucket=bucket, write_precision=args.precision, record=batch_lines)
print(f'Wrote {len(batch_lines)} lines')
sleep(1) # sleep for 1 second avoid rate limit

Ref Python Doc - https://docs.python.org/3/library/itertools.html#itertools.batched

I realize that there may be more than one solutions in pure Python, and you can choose the best one. Let me start with this:

def batched(iterable: Iterable[T], n: int) -> Iterator[tuple[T, ...]]:
    iter_ = iter(iterable)
    while True:
        batch = tuple(itertools.islice(iter_, n))
        if not batch:
            break
        yield batch

To make it shorter(but not recommended):

def batched(iterable: Iterable[T], n: int) -> Iterator[tuple[T, ...]]:
    iter_ = iter(iterable)
    return iter(lambda: tuple(itertools.islice(iter_, n)), ())

@frostming Let's rocks :D

Would you send a patch directly or I'd do you a favor to convey this function ...

BTW, I just noticed that I don't add type checks here. And I ever thought of using PDM to build venv and deps but I rush into a working demo first.