Parse Records in Batches with a Generator

A job reads a long stream of name:count lines, such as alpha:7, and sends the records on in batches. The stream can be too big for a list, so parse it lazily. Read a line only when the caller asks for the batch that needs it. The supplied CountingSource counts every line your code reads, so the tests can check when the reading happens.

batches = parse_records(["a:1", "b:2", "c:3"], chunk=2)  # reads nothing yet
next(batches)  # [Record(name='a', count=1), Record(name='b', count=2)]
next(batches)  # [Record(name='c', count=3)]

Write parse_line

parse_line(line) turns one line into a Record.

• Split the line at its first :. The part before it is the name, and the part after it is the count.
• The name must not be empty. Keep it exactly as written, spaces included.
• The count must pass str.isdecimal(). That allows leading zeros ("alpha:007" gives 7) and decimal digits from other writing systems ("x:٣" gives 3). It rejects an empty count, signs, spaces, decimal points, a second colon and superscripts such as "x:²".
• Convert the count with int(). If int() still raises, for example on a count thousands of digits long, raise the same error as for any bad line.
• For any bad line, raise ValueError(f"malformed record: {line!r}"). For example, ":7", "x:-1" and "x:1:2" are all malformed.

Write parse_records

parse_records(lines, *, chunk) is a generator function. It yields lists of records in input order.

• Calling it must not read any line.
• If chunk <= 0, raise ValueError(f"chunk must be positive, got {chunk}") before you read any line. Put this check inside the generator body. It then runs at the first next(), not at the call. This is the opposite of bulk_pages in T5, and it means a caller who never asks for a batch never sees the error.
• Yield a batch as soon as it holds chunk records. Read no line beyond the batch the caller asked for.
• When the lines run out, yield the last short batch if it holds any records. An empty source yields no batches, not one empty batch.
• Start a new list for each batch.
• When parse_line raises on a bad line, let its ValueError end the generator. Batches you already yielded stay delivered. The unfinished batch never reaches the caller.

Write solve

solve runs parse_records over the lines and reports what happened.

1. Wrap the lines in a CountingSource. Pass lines itself, or itertools.cycle(lines) when repeat is true. cycle repeats the lines forever, so the tests always pass a take when repeat is true. An empty list stays empty under cycle.
2. Call parse_records(source, chunk=chunk), and record source.pulled right away as pulled_before_first.
3. If take is not None, read at most take batches, using itertools.islice(batches, take). With take=0, islice never calls next(), so the generator body never runs. That is why chunk=0, take=0 gives no error.
4. Loop over the batches inside one try block. Catch ValueError and store str(exc) as error.

solve returns these keys:

If no batch arrives, batch_sizes and head are empty and total is 0.

For lines=["a:1", "b:2", "bad"] and chunk=2, you get one batch of 2, total 3, pulled 3 and the error "malformed record: 'bad'". Add take=1, and your code never reads the bad line. Then pulled is 2 and there is no error. With repeat=True, chunk=2 and take=3, valid lines give three batches after exactly six reads.