Boss level: the data pipeline

What you'll learn

Use generators, decorators and type hints together
Skip faulty data robustly
Write professional, readable code

The final boss

You’ve almost finished the pro stage: generators, decorators, type hints, async. In the last boss level you build a data pipeline – that’s what a chain of processing steps is called, and they’re everywhere in science and industry: raw data in, clean results out.

Building block 1: a generator with a filter

def only_even(numbers):
    for n in numbers:
        if n % 2 == 0:
            yield n


print(list(only_even(range(10))))

A generator calculates on demand – even with millions of readings it hardly needs any memory.

Building block 2: a decorator with a memory

Functions are objects – you can even give them attributes:

import functools


def loud(function):
    @functools.wraps(function)
    def wrapper(*args):
        print("→ calling", function.__name__)
        return function(*args)

    return wrapper


@loud
def double(x: int) -> int:
    return 2 * x


print(double(21))
print(double.__name__)   # thanks to functools.wraps not “wrapper”

Building block 3: stay robust

Real data is never perfect. Instead of crashing, good software skips broken entries – and maybe counts them:

inputs = ["3", "three", "4"]
for i in inputs:
    try:
        print(int(i) * 10)
    except ValueError:
        print("skipped:", repr(i))

Done it? Then you’ve mastered all three stages – respect! Next: take on a project of your own. The playground is your lab. 🎓

Your task

Your turn

A sensor delivers readings as text – some are broken ("Error", ""). Write the generator readings(lines) that only yields the valid values as float (skip broken ones with try/except). Write the decorator count_calls that counts in function.calls how often the function was called. Decorate mean(values: list[float]) -> float with it; it returns the average.

def readings(lines):
    pass


def count_calls(function):
    pass


def mean(values):
    pass


data = ["21.5", "Error", "22.0", "", "23.5"]
values = list(readings(data))
print(values)
print(mean(values))

Something to think about
A generator hands out values with yield instead of returning them all with return.
Which command helps?
float("x") raises a ValueError. Catch it with try/except and carry on with continue.
The wrapper function in the decorator increases wrapper.calls by 1 and then calls the real function. Set wrapper.calls = 0 before returning wrapper – and write @count_calls above mean.
Add the types: def mean(values: list[float]) -> float: – and use functools.wraps in the decorator so they are kept.

Gap code – fill in the ___

                        ___


def readings(lines):
    for line in ___:
        try:
            ___
        except ValueError:
            continue


def count_calls(function):
    ___
    def wrapper(*args, **kwargs):
        wrapper.calls += ___
        return ___

    wrapper.calls = ___
    return ___


___
def mean(values: list[float]) -> float:
    return ___


data = ["21.5", "Error", "22.0", "", "23.5"]
values = list(readings(data))
print(values)
print(mean(values))

Solve it without hints to earn a ⭐ on your progress page.

Show solution

Try it yourself first – mistakes teach you the most. Here is one possible solution:

import functools


def readings(lines):
    for line in lines:
        try:
            yield float(line)
        except ValueError:
            continue


def count_calls(function):
    @functools.wraps(function)
    def wrapper(*args, **kwargs):
        wrapper.calls += 1
        return function(*args, **kwargs)

    wrapper.calls = 0
    return wrapper


@count_calls
def mean(values: list[float]) -> float:
    return sum(values) / len(values)


data = ["21.5", "Error", "22.0", "", "23.5"]
values = list(readings(data))
print(values)
print(mean(values))