Build a Streaming Pipeline from Start to Finish

This last problem of the track combines chain, islice, batched and cycle. Your service reads key=count lines from several sources. You want the total per key for each small batch of lines, and you want to stop after a set number of lines. It is a bigger version of Build a Pipeline from Small Generators. Every stage is lazy, so no stage reads a line until the caller asks for a result.

list(process([["a=1", "b=2"], ["a=3"]], batch=2, limit=10))
# [{"a": 1, "b": 2}, {"a": 3}]

list(process([["a=1", "b=2"], ["a=3"]], batch=10, limit=10))
# [{"a": 4, "b": 2}]

Each batch gets its own totals. A key that shows up in two batches has a separate total in each.

Write the stages

Use the signatures in the starter. Keep LineCounter and Report as they are.

• flatten(sources) joins the inner iterables into one stream of lines. It finishes each inner iterable before it starts the next, skips empty ones, and reads nothing when called. chain.from_iterable(sources) does all of this.
• parse(lines) yields (key, count) for each line, one line at a time. Split the line at the first =. Keep the key exactly as it is, even if it is empty or has spaces.
• Convert the count with plain int(). It accepts a sign, spaces around the number, and decimal digits from any writing system, so "a=٣" gives ("a", 3).
• A line with no =, or a count that int() rejects, raises ValueError when parse reaches it. Let that error reach the caller, and do not skip the line. Any extra = stays in the count, so "a=1=2" raises too.
• aggregate(pairs) adds up the counts for each key. It returns a dict with the keys in sorted order, and {} when there are no pairs. Leave the pairs you get unchanged.
• process(sources, *, batch, limit) flattens the sources, takes at most limit lines, and parses them. It groups the pairs into batches of batch, and yields one aggregate dict per batch. The last batch can be short. Never yield an empty batch.
• process reads nothing when called. With limit=0 it yields nothing and reads no lines, even from a source that never ends.

batch is always at least 1, and limit is at least 0.

Put the limit on lines

limit counts lines across all the sources, wherever one source ends and the next begins. Apply islice to the flattened lines, before you parse and batch them. On the outer iterable, islice would count sources. On the output, it would count batches. In the first example above, limit=2 gives only {"a": 1, "b": 2}.

When a line fails to parse, process raises and drops the batch it was building. Batches it already yielded stay with the caller. solve does not catch the error. The tests use only valid lines.

Write solve

• Use sources as the outer iterable. When repeat is True, repeat it forever with itertools.cycle, but only if at least one inner list has a line. If every inner list is empty, keep the plain list. Cycling empty lists would loop forever without producing a line for islice to count.
• Wrap each inner list with counter.wrap(...), using one shared LineCounter. Wrap them lazily, with a generator expression, because the cycled outer iterable never ends.
• Call process, then read counter.pulled before you take any result. Then collect every batch into a list.

The report has three keys:

• batches holds the batch dicts in the order process yielded them.
• pulled is the number of lines the sources handed out.
• pulled_before_first is counter.pulled right after process returned. It must be 0.

For finite input, pulled is limit or the number of lines, whichever is smaller. For repeated input with at least one line, it equals limit. Input with no lines, including repeated empty lists, gives no batches and a pulled of 0.

Memory

At any moment, batched holds up to batch pairs, and aggregate holds one total per key in that batch. Sorting stays cheap, because aggregate sorts one batch's keys. Calling sorted on the whole input would load all of it. The report itself keeps every batch, so it grows with the number of batches you collect.

All your functions must pass mypy --strict.