Build a lazy generator pipeline · Python practice

Build a lazy generator pipeline

Medium~20 min

Write pipeline(source, n) returning a generator that, from an iterator of log records ({"level": str, "ms": int}):

keeps records with level "ERROR",
maps each to its ms value,
yields lists (batches) of up to 3 values,
stops after yielding n batches.

It must be lazy: it should pull from source only as many records as needed. The harness counts how many records you consumed.

Function

Implement pipeline. Input: source: iterator of dict, n: int. Returns: generator of list[int].

Examples

>>> pipeline([{'ms': 1, 'level': 'ERROR'}, {'ms': 2, 'level': 'INFO'}, {'ms': 3, 'level': 'ERROR'}, {'ms': 4, 'level': 'ERROR'}, {'ms': 5, 'level': 'ERROR'}, {'ms': 6, 'level': 'ERROR'}], 1)

{'lazy': True, 'pulled': 4, 'batches': [[1, 3, 4]]}

The first batch is complete after the fourth record, so the last two are never read.

Provided environment

This code runs before yours (fake clients, clocks and the test harness). You can use its names; don't edit them.

def __harness__(fn, records, n):
    pulled = {"count": 0}

    def source():
        for r in records:
            pulled["count"] += 1
            yield r

    gen = fn(source(), n)
    is_generator = hasattr(gen, "__next__")
    batches = list(gen)
    return {"batches": batches, "pulled": pulled["count"], "lazy": is_generator}

Details

Level: API integration against fakes, data processing, error handling, decorators and asyncio. Several cooperating parts.
Roles: Python Developer, Data Engineer, Machine Learning Engineer
Topics: Generators and laziness, Time and space complexity
Tests: 1 public (run with Run), 3 hidden (graded on Submit)
Grading: Exact match of the return value