// I3.6 · ca. 40 Min. · Aufbau

Projekt: ein lazy CSV-Bericht mit collections

In diesem Projekt lesen Sie eine Bestelldatei lazy mit csv und einem Generator, zählen mit Counter, gruppieren mit defaultdict, benennen Felder mit namedtuple und behalten die neuesten Bestellungen mit deque.

Lektion 6 von 6 in I3 Iteration und funktionale Werkzeuge

ZurückEnde des Moduls

Danach können Sie

Mit Counter zählen und mit defaultdict gruppieren, statt fehlende Schlüssel zu prüfen
Felder eines Datensatzes mit namedtuple benennen und die letzten n Elemente mit deque(maxlen=n) behalten
Eine CSV-Datei lazy mit csv.DictReader und einem Generator lesen und durch eine Pipeline schicken

Aufwärmen · Aufgabe 1 von 7
Aufwärmen aus Lektion I3.2: Einen Generator können Sie einmal durchlaufen. Was gibt das aus?
```
g = (n for n in [1, 2])
print(list(g), list(g))
```
[1, 2] [1, 2][] [][1, 2] []StopIteration
Vorhersagen · Aufgabe 2 von 7
Sagen Sie voraus, bevor Sie weiterlesen: z kommt in "banana" nicht vor. Was gibt das aus?
```
from collections import Counter

c = Counter("banana")
print(c["a"], c["z"])
```
3 None1 0KeyError: 'z'3 0
Üben · Aufgabe 3 von 7
Ergänzen Sie die Factory, damit jeder neue Anfangsbuchstabe mit einer leeren Liste beginnt.
```
from collections import defaultdict

groups = defaultdict(____)
for word in ["apple", "bean", "avocado"]:
    groups[word[0]].append(word)
print(dict(groups))
```
groups = defaultdict(Ihre Antwort)
Üben · Aufgabe 4 von 7
Fünf Zahlen kommen in eine deque, die drei fasst. Was gibt das aus?
```
from collections import deque

d = deque(maxlen=3)
for n in range(1, 6):
    d.append(n)
print(list(d))
```
[1, 2, 3][5][1, 2, 3, 4, 5][3, 4, 5]
Üben · Aufgabe 5 von 7
Counter, defaultdict, deque und namedtuple sind aus collections importiert. Ordnen Sie jedem Ausdruck seinen Wert zu.
Counter("abca")["a"] — Zuordnung für „Counter("abca")["a"]“
Counter("abca").most_common(1) — Zuordnung für „Counter("abca").most_common(1)“
"".join(deque("abcde", maxlen=2)) — Zuordnung für „"".join(deque("abcde", maxlen=2))“
defaultdict(list)["x"] — Zuordnung für „defaultdict(list)["x"]“
namedtuple("P", "x y")(3, 4).y — Zuordnung für „namedtuple("P", "x y")(3, 4).y“
Denksport · Aufgabe 6 von 7
Knobelaufgabe. Die deque behält nur ein Element. Was gibt das aus?
```
from collections import deque

def numbers():
    for n in range(1, 4):
        print("made", n, end="; ")
        yield n

print(deque(numbers(), maxlen=1))
```
made 3; deque([3], maxlen=1)made 1; made 2; made 3; deque([3], maxlen=1)made 1; made 2; made 3; deque([1, 2, 3], maxlen=1)deque([3], maxlen=1)
Anwenden · Aufgabe 7 von 7
Kleine Aufgabe. LOG ist eine Liste von CSV-Zeilen mit der Kopfzeile level,message. Schreiben Sie einen Generator records(lines), der mit csv.DictReader Paare (level, message) liefert. Geben Sie dann mit Counter aus, wie oft jedes Level vorkommt, das häufigste zuerst, und mit einer deque mit maxlen=2 die letzten zwei ERROR-Meldungen.
Prüfen Sie Ihr Ergebnis anhand dieser Liste
- records ist ein Generator, der csv.DictReader nutzt und (level, message) liefert
- Die Level-Zählung kommt aus Counter(...).most_common()
- Die letzten zwei Fehler kommen aus deque(..., maxlen=2) über einem Generatorausdruck
- Ausgabe: [('ERROR', 3), ('INFO', 2), ('WARN', 1)] und dann ['timeout', 'disk full']

Selbst programmieren

Lesen Sie das ausgearbeitete Beispiel und lösen Sie dann die Übungen. Ihr Code läuft in Ihrem Browser oder auf Ihrem Computer und wird nie hochgeladen.

Ausgearbeitetes Beispiel

Seitenaufrufe, gezählt und gruppiert

Das Besuchsprotokoll ist eine Liste von CSV-Zeilen, das Beispiel läuft also ohne Datei; csv.DictReader nimmt jedes Iterable von Strings, auch eine geöffnete Datei. visits() ist ein Generator, der jede Zeile in ein Visit-namedtuple verwandelt. Jeder Verbraucher ruft visits(LINES) für einen frischen Durchlauf neu auf: Counter zählt Seiten, ein defaultdict gruppiert Seiten nach Nutzer, und eine begrenzte deque behält die zwei neuesten Besuche.

main.py

import csv
from collections import Counter, defaultdict, deque, namedtuple
from collections.abc import Iterable, Iterator

Visit = namedtuple("Visit", ["page", "user", "seconds"])

LINES = [
    "page,user,seconds",
    "home,ada,5",
    "docs,bo,30",
    "home,bo,3",
    "blog,ada,12",
    "docs,ada,8",
    "home,cy,4",
]


def visits(lines: Iterable[str]) -> Iterator[Visit]:
    """Yield one Visit per CSV row, converting seconds to int."""
    for row in csv.DictReader(lines):
        yield Visit(row["page"], row["user"], int(row["seconds"]))


first = next(visits(LINES))
print(first, first.page, first[2])

views = Counter(v.page for v in visits(LINES))
print(views.most_common(2), views["shop"])

pages_by_user: defaultdict[str, list[str]] = defaultdict(list)
for v in visits(LINES):
    pages_by_user[v.user].append(v.page)
print(dict(pages_by_user))

recent = deque(visits(LINES), maxlen=2)
print([v.page for v in recent])

Ausführen mit

python main.py

Ausgabe

Visit(page='home', user='ada', seconds=5) home 5
[('home', 3), ('docs', 2)] 0
{'ada': ['home', 'blog', 'docs'], 'bo': ['docs', 'home'], 'cy': ['home']}
['docs', 'home']

first.page und first[2] lesen dasselbe Tupel über Namen und über Position.
views["shop"] ist 0: Ein Counter löst für ein fehlendes Element keinen KeyError aus.
Jeder Nutzer bekam beim ersten Auftreten eine frische Liste, ohne Prüfung mit if user not in ....
Die deque hat alle sechs Besuche gelesen, aber nur die letzten zwei behalten.

Ändern und ausführen

Ihr Code (main.py)

Tab rückt ein, Umschalt+Tab rückt aus. Um den Editor mit der Tastatur zu verlassen, drücken Sie Esc und dann Tab.

Beim ersten Ausführen lädt Ihr Browser Python herunter (bis zu 6.5 MB) und speichert es im Cache. Ihr Code bleibt auf Ihrem Gerät.

Übungen

Übung 1 von 4

Schritt 1: Bestellungen lazy lesen

orders.csv (zweiter Tab) hat die Kopfzeile customer,product,qty. Der Starter trennt jede Zeile an Kommas, was am Produkt "tea, green" in Anführungszeichen scheitert, lässt qty als String und baut eine ganze Liste. Schreiben Sie read_orders(path) als Generator: Öffnen Sie die Datei mit newline="", lesen Sie sie mit csv.DictReader und liefern Sie pro Zeile eine Order, mit qty als int.

Ihr Code (main.py)

Tab rückt ein, Umschalt+Tab rückt aus. Um den Editor mit der Tastatur zu verlassen, drücken Sie Esc und dann Tab.

Beim ersten Ausführen lädt Ihr Browser Python herunter (bis zu 6.5 MB) und speichert es im Cache. Ihr Code bleibt auf Ihrem Gerät.

Hinweise

Hinweis 1
Im with-Block: for row in csv.DictReader(f):. Jede Zeile ist ein Dict mit der Kopfzeile als Schlüsseln, row["qty"] ist also die Menge.
Hinweis 2
Ersetzen Sie orders.append(...) durch yield Order(...), streichen Sie die Liste und ändern Sie den Rückgabetyp in Iterator[Order].
Hinweis 3
yield Order(row["customer"], row["product"], int(row["qty"])). DictReader liest die Kopfzeile selbst, next(f) muss also weg.

Eine Lösung zeigen

Ein möglicher Lösungsweg. Ihrer kann anders aussehen und trotzdem alle Prüfungen bestehen.

import csv
from collections import Counter, defaultdict, deque, namedtuple
from collections.abc import Iterable, Iterator

Order = namedtuple("Order", ["customer", "product", "qty"])


def read_orders(path: str) -> Iterator[Order]:
    """Yield one Order per row of the CSV file at path, lazily."""
    with open(path, encoding="utf-8", newline="") as f:
        for row in csv.DictReader(f):
            yield Order(row["customer"], row["product"], int(row["qty"]))


if __name__ == "__main__":
    for order in read_orders("orders.csv"):
        print(order)

Auf dem eigenen Computer ausführen

Installieren Sie Python 3.14 oder neuer. Speichern Sie diese Dateien in einem Ordner, öffnen Sie dort ein Terminal und führen Sie die Befehle unten aus.

main.py

import csv
from collections import Counter, defaultdict, deque, namedtuple
from collections.abc import Iterable, Iterator

Order = namedtuple("Order", ["customer", "product", "qty"])


def read_orders(path: str) -> list[Order]:
    """Read every row of the CSV file at path into a list of Orders."""
    orders: list[Order] = []
    with open(path, encoding="utf-8") as f:
        next(f)  # skip the header line
        for line in f:
            customer, product, qty = line.strip().split(",")
            orders.append(Order(customer, product, qty))
    return orders


if __name__ == "__main__":
    for order in read_orders("orders.csv"):
        print(order)

test_main.py

from main import Order, read_orders


def test_first_order():
    """Die erste Bestellung ist ada, tea, 2"""
    got = next(iter(read_orders("orders.csv")))
    assert got == Order("ada", "tea", 2), f"die erste Bestellung ist {got!r}, erwartet: ada, tea und die Zahl 2"


def test_qty_is_int():
    """qty wird in int umgewandelt"""
    got = [order.qty for order in read_orders("orders.csv")]
    assert got == [2, 1, 3, 1, 2, 1], f"die Mengen sind {got!r}, erwartet: die ints [2, 1, 3, 1, 2, 1]"


def test_quoted_comma():
    """Ein Produkt in Anführungszeichen darf ein Komma enthalten"""
    got = list(read_orders("orders.csv"))[4]
    assert got == Order("bo", "tea, green", 2), f"die fünfte Bestellung ist {got!r}, erwartet: bo, tea, green und 2"


def test_lazy():
    """read_orders ist ein Generator und liefert einen Iterator"""
    orders = read_orders("orders.csv")
    assert iter(orders) is orders, f"read_orders lieferte ein {type(orders).__name__}: Nutzen Sie yield statt eine Liste zu bauen"


def test_other_file():
    """Eine andere Datei ergibt ihre eigenen Bestellungen"""
    with open("other.csv", "w", encoding="utf-8", newline="") as f:
        f.write("customer,product,qty\nzoe,jam,5\n")
    got = list(read_orders("other.csv"))
    assert got == [Order("zoe", "jam", 5)], f"für other.csv lieferte read_orders {got!r}"

orders.csv

customer,product,qty
ada,tea,2
bo,coffee,1
ada,coffee,3
cy,tea,1
bo,"tea, green",2
ada,cake,1

Unter macOS und Linux tippen Sie python3, wo in diesen Befehlen python steht, wie in der ersten Lektion.

Programm ausführen:

python main.py

Prüfungen ausführen (learnrun.py muss im selben Ordner liegen):

python learnrun.py test

learnrun.py herunterladen

Übung 2 von 4

Schritt 2: Mengen mit Counter zählen

units_by_product(orders) soll die qty aller Bestellungen je Produkt addieren und einen Counter zurückgeben, damit most_common() funktioniert und ein nie bestelltes Produkt 0 zählt. Der Starter nutzt ein normales Dict und überschreibt sogar frühere Bestellungen desselben Produkts. Für orders.csv ergibt sich coffee 4, tea 3, tea, green 2 und cake 1.

Ihr Code (main.py)

Tab rückt ein, Umschalt+Tab rückt aus. Um den Editor mit der Tastatur zu verlassen, drücken Sie Esc und dann Tab.

Beim ersten Ausführen lädt Ihr Browser Python herunter (bis zu 6.5 MB) und speichert es im Cache. Ihr Code bleibt auf Ihrem Gerät.

Hinweise

Hinweis 1
Beginnen Sie mit units: Counter[str] = Counter() und ändern Sie den Rückgabetyp in Counter[str].
Hinweis 2
Addieren statt zuweisen: units[order.product] += order.qty. Ein Counter beginnt jedes neue Produkt bei 0.
Hinweis 3
Counter ist oben in main.py schon aus collections importiert.

Eine Lösung zeigen

Ein möglicher Lösungsweg. Ihrer kann anders aussehen und trotzdem alle Prüfungen bestehen.

import csv
from collections import Counter, defaultdict, deque, namedtuple
from collections.abc import Iterable, Iterator

Order = namedtuple("Order", ["customer", "product", "qty"])


def read_orders(path: str) -> Iterator[Order]:
    """Yield one Order per row of the CSV file at path, lazily."""
    with open(path, encoding="utf-8", newline="") as f:
        for row in csv.DictReader(f):
            yield Order(row["customer"], row["product"], int(row["qty"]))


def units_by_product(orders: Iterable[Order]) -> Counter[str]:
    """Count the units ordered of each product."""
    units: Counter[str] = Counter()
    for order in orders:
        units[order.product] += order.qty
    return units


if __name__ == "__main__":
    print(units_by_product(read_orders("orders.csv")))

Auf dem eigenen Computer ausführen

Installieren Sie Python 3.14 oder neuer. Speichern Sie diese Dateien in einem Ordner, öffnen Sie dort ein Terminal und führen Sie die Befehle unten aus.

main.py

import csv
from collections import Counter, defaultdict, deque, namedtuple
from collections.abc import Iterable, Iterator

Order = namedtuple("Order", ["customer", "product", "qty"])


def read_orders(path: str) -> Iterator[Order]:
    """Yield one Order per row of the CSV file at path, lazily."""
    with open(path, encoding="utf-8", newline="") as f:
        for row in csv.DictReader(f):
            yield Order(row["customer"], row["product"], int(row["qty"]))


def units_by_product(orders: Iterable[Order]) -> dict[str, int]:
    """Count the units ordered of each product."""
    units: dict[str, int] = {}
    for order in orders:
        units[order.product] = order.qty
    return units


if __name__ == "__main__":
    print(units_by_product(read_orders("orders.csv")))

test_main.py

from collections import Counter

from main import Order, read_orders, units_by_product


def test_given_file():
    """orders.csv ergibt coffee 4, tea 3, tea, green 2, cake 1"""
    got = units_by_product(read_orders("orders.csv"))
    assert got == {"coffee": 4, "tea": 3, "tea, green": 2, "cake": 1}, f"units_by_product lieferte {got!r}"


def test_adds_up():
    """Bestellungen desselben Produkts werden addiert"""
    got = units_by_product([Order("a", "jam", 2), Order("b", "jam", 5)])
    jam = got["jam"]
    assert jam == 7, f"jam hat {jam!r} Einheiten, erwartet: 2 + 5 = 7"


def test_is_counter():
    """Das Ergebnis ist ein Counter mit most_common und 0 für fehlende Produkte"""
    got = units_by_product([Order("a", "jam", 2), Order("b", "tea", 1), Order("c", "jam", 1)])
    assert isinstance(got, Counter), f"units_by_product lieferte ein {type(got).__name__}, erwartet: einen Counter"
    top = got.most_common(1)
    assert top == [("jam", 3)], f"most_common(1) ist {top!r}, erwartet: [('jam', 3)]"
    assert got["cake"] == 0, "ein nie bestelltes Produkt soll 0 zählen"


def test_generator_input():
    """Auch ein Generator von Bestellungen funktioniert"""
    got = units_by_product(Order("a", "p", n) for n in range(4))
    assert got == {"p": 6}, f"units_by_product lieferte {got!r}, erwartet: p mit 0 + 1 + 2 + 3 = 6"

orders.csv

customer,product,qty
ada,tea,2
bo,coffee,1
ada,coffee,3
cy,tea,1
bo,"tea, green",2
ada,cake,1

Unter macOS und Linux tippen Sie python3, wo in diesen Befehlen python steht, wie in der ersten Lektion.

Programm ausführen:

python main.py

Prüfungen ausführen (learnrun.py muss im selben Ordner liegen):

python learnrun.py test

learnrun.py herunterladen

Übung 3 von 4

Schritt 3: Produkte mit defaultdict gruppieren

products_by_customer(orders) soll jedem Kunden die Menge der verschiedenen Produkte zuordnen, die er bestellt hat. Der Starter ersetzt die Menge bei jeder Bestellung, sodass nur das letzte Produkt übrig bleibt. Nutzen Sie defaultdict(set), fügen Sie jedes Produkt hinzu und geben Sie mit dict(...) ein normales Dict zurück. Für orders.csv hat ada tea, coffee und cake bestellt.

Ihr Code (main.py)

Tab rückt ein, Umschalt+Tab rückt aus. Um den Editor mit der Tastatur zu verlassen, drücken Sie Esc und dann Tab.

import csv
from collections import Counter, defaultdict, deque, namedtuple
from collections.abc import Iterable, Iterator

Order = namedtuple("Order", ["customer", "product", "qty"])

def read_orders(path: str) -> Iterator[Order]:
    """Yield one Order per row of the CSV file at path, lazily."""
    with open(path, encoding="utf-8", newline="") as f:
        for row in csv.DictReader(f):
            yield Order(row["customer"], row["product"], int(row["qty"]))

def units_by_product(orders: Iterable[Order]) -> Counter[str]:
    """Count the units ordered of each product."""
    units: Counter[str] = Counter()
    for order in orders:
        units[order.product] += order.qty
    return units

def products_by_customer(orders: Iterable[Order]) -> dict[str, set[str]]:
    """Map each customer to the set of products they ordered."""
    products: dict[str, set[str]] = {}
    for order in orders:
        products[order.customer] = {order.product}
    return products

if __name__ == "__main__":
    print(products_by_customer(read_orders("orders.csv")))

Beim ersten Ausführen lädt Ihr Browser Python herunter (bis zu 6.5 MB) und speichert es im Cache. Ihr Code bleibt auf Ihrem Gerät.

Hinweise

Hinweis 1
products: defaultdict[str, set[str]] = defaultdict(set) gibt jedem neuen Kunden eine leere Menge.
Hinweis 2
products[order.customer].add(order.product) fügt zu dieser Menge hinzu; eine Menge ignoriert ein Produkt, das sie schon hat.
Hinweis 3
Enden Sie mit return dict(products), damit der Aufrufer ein normales Dict bekommt.

Eine Lösung zeigen

Ein möglicher Lösungsweg. Ihrer kann anders aussehen und trotzdem alle Prüfungen bestehen.

import csv
from collections import Counter, defaultdict, deque, namedtuple
from collections.abc import Iterable, Iterator

Order = namedtuple("Order", ["customer", "product", "qty"])


def read_orders(path: str) -> Iterator[Order]:
    """Yield one Order per row of the CSV file at path, lazily."""
    with open(path, encoding="utf-8", newline="") as f:
        for row in csv.DictReader(f):
            yield Order(row["customer"], row["product"], int(row["qty"]))


def units_by_product(orders: Iterable[Order]) -> Counter[str]:
    """Count the units ordered of each product."""
    units: Counter[str] = Counter()
    for order in orders:
        units[order.product] += order.qty
    return units


def products_by_customer(orders: Iterable[Order]) -> dict[str, set[str]]:
    """Map each customer to the set of products they ordered."""
    products: defaultdict[str, set[str]] = defaultdict(set)
    for order in orders:
        products[order.customer].add(order.product)
    return dict(products)


if __name__ == "__main__":
    print(products_by_customer(read_orders("orders.csv")))

Auf dem eigenen Computer ausführen

Installieren Sie Python 3.14 oder neuer. Speichern Sie diese Dateien in einem Ordner, öffnen Sie dort ein Terminal und führen Sie die Befehle unten aus.

main.py

import csv
from collections import Counter, defaultdict, deque, namedtuple
from collections.abc import Iterable, Iterator

Order = namedtuple("Order", ["customer", "product", "qty"])


def read_orders(path: str) -> Iterator[Order]:
    """Yield one Order per row of the CSV file at path, lazily."""
    with open(path, encoding="utf-8", newline="") as f:
        for row in csv.DictReader(f):
            yield Order(row["customer"], row["product"], int(row["qty"]))


def units_by_product(orders: Iterable[Order]) -> Counter[str]:
    """Count the units ordered of each product."""
    units: Counter[str] = Counter()
    for order in orders:
        units[order.product] += order.qty
    return units


def products_by_customer(orders: Iterable[Order]) -> dict[str, set[str]]:
    """Map each customer to the set of products they ordered."""
    products: dict[str, set[str]] = {}
    for order in orders:
        products[order.customer] = {order.product}
    return products


if __name__ == "__main__":
    print(products_by_customer(read_orders("orders.csv")))

test_main.py

from main import Order, products_by_customer, read_orders


def test_given_file():
    """orders.csv: ada hat tea, coffee und cake bestellt"""
    got = products_by_customer(read_orders("orders.csv"))
    expected = {"ada": {"tea", "coffee", "cake"}, "bo": {"coffee", "tea, green"}, "cy": {"tea"}}
    assert got == expected, f"products_by_customer lieferte {got!r}, erwartet: {expected!r}"


def test_no_duplicates():
    """Ein zweimal bestelltes Produkt erscheint einmal"""
    got = products_by_customer([Order("a", "jam", 1), Order("a", "jam", 2)])
    assert got == {"a": {"jam"}}, f"products_by_customer lieferte {got!r}, erwartet: a mit der Menge aus jam"


def test_empty():
    """Ohne Bestellungen kommt ein leeres Dict"""
    got = products_by_customer([])
    assert got == {}, f"products_by_customer([]) lieferte {got!r}, erwartet: {{}}"


def test_plain_dict():
    """Das Ergebnis ist ein normales Dict"""
    got = products_by_customer([Order("a", "jam", 1)])
    assert type(got) is dict, f"products_by_customer lieferte ein {type(got).__name__}: Wandeln Sie es mit dict(...) um"

orders.csv

customer,product,qty
ada,tea,2
bo,coffee,1
ada,coffee,3
cy,tea,1
bo,"tea, green",2
ada,cake,1

Unter macOS und Linux tippen Sie python3, wo in diesen Befehlen python steht, wie in der ersten Lektion.

Programm ausführen:

python main.py

Prüfungen ausführen (learnrun.py muss im selben Ordner liegen):

python learnrun.py test

learnrun.py herunterladen

Übung 4 von 4

Schritt 4: die neuesten Bestellungen und der Bericht

last_orders(orders, n) liefert die letzten n Bestellungen. Der Starter macht aus allen Bestellungen eine Liste und liefert bei n = 0 sogar alle, weil [-0:] die ganze Liste ist. Nutzen Sie deque(orders, maxlen=n), das nur n Bestellungen im Speicher hält. Der Bericht unten in main.py gibt dann die zwei meistverkauften Produkte, die Produkte von ada und die letzten zwei Bestellungen aus.

Ihr Code (main.py)

Tab rückt ein, Umschalt+Tab rückt aus. Um den Editor mit der Tastatur zu verlassen, drücken Sie Esc und dann Tab.

import csv
from collections import Counter, defaultdict, deque, namedtuple
from collections.abc import Iterable, Iterator

Order = namedtuple("Order", ["customer", "product", "qty"])

def products_by_customer(orders: Iterable[Order]) -> dict[str, set[str]]:
    """Map each customer to the set of products they ordered."""
    products: defaultdict[str, set[str]] = defaultdict(set)
    for order in orders:
        products[order.customer].add(order.product)
    return dict(products)

def last_orders(orders: Iterable[Order], n: int) -> list[Order]:
    """Return the last n orders, reading the orders only once."""
    return list(orders)[-n:]

if __name__ == "__main__":
    print(units_by_product(read_orders("orders.csv")).most_common(2))
    print(sorted(products_by_customer(read_orders("orders.csv"))["ada"]))
    for order in last_orders(read_orders("orders.csv"), 2):
        print(order.customer, order.product, order.qty)

Beim ersten Ausführen lädt Ihr Browser Python herunter (bis zu 6.5 MB) und speichert es im Cache. Ihr Code bleibt auf Ihrem Gerät.

Hinweise

Hinweis 1
deque(orders, maxlen=n) liest jede Bestellung, behält aber nur die letzten n.
Hinweis 2
Geben Sie wie im Typ angegeben eine Liste zurück: list(deque(orders, maxlen=n)).
Hinweis 3
Eine deque mit maxlen=0 behält nichts, n = 0 ergibt also [] ohne Sonderfall.

Eine Lösung zeigen

Ein möglicher Lösungsweg. Ihrer kann anders aussehen und trotzdem alle Prüfungen bestehen.

import csv
from collections import Counter, defaultdict, deque, namedtuple
from collections.abc import Iterable, Iterator

Order = namedtuple("Order", ["customer", "product", "qty"])


def read_orders(path: str) -> Iterator[Order]:
    """Yield one Order per row of the CSV file at path, lazily."""
    with open(path, encoding="utf-8", newline="") as f:
        for row in csv.DictReader(f):
            yield Order(row["customer"], row["product"], int(row["qty"]))


def units_by_product(orders: Iterable[Order]) -> Counter[str]:
    """Count the units ordered of each product."""
    units: Counter[str] = Counter()
    for order in orders:
        units[order.product] += order.qty
    return units


def products_by_customer(orders: Iterable[Order]) -> dict[str, set[str]]:
    """Map each customer to the set of products they ordered."""
    products: defaultdict[str, set[str]] = defaultdict(set)
    for order in orders:
        products[order.customer].add(order.product)
    return dict(products)


def last_orders(orders: Iterable[Order], n: int) -> list[Order]:
    """Return the last n orders, reading the orders only once."""
    return list(deque(orders, maxlen=n))


if __name__ == "__main__":
    print(units_by_product(read_orders("orders.csv")).most_common(2))
    print(sorted(products_by_customer(read_orders("orders.csv"))["ada"]))
    for order in last_orders(read_orders("orders.csv"), 2):
        print(order.customer, order.product, order.qty)

Auf dem eigenen Computer ausführen

Installieren Sie Python 3.14 oder neuer. Speichern Sie diese Dateien in einem Ordner, öffnen Sie dort ein Terminal und führen Sie die Befehle unten aus.

main.py

import csv
from collections import Counter, defaultdict, deque, namedtuple
from collections.abc import Iterable, Iterator

Order = namedtuple("Order", ["customer", "product", "qty"])


def read_orders(path: str) -> Iterator[Order]:
    """Yield one Order per row of the CSV file at path, lazily."""
    with open(path, encoding="utf-8", newline="") as f:
        for row in csv.DictReader(f):
            yield Order(row["customer"], row["product"], int(row["qty"]))


def units_by_product(orders: Iterable[Order]) -> Counter[str]:
    """Count the units ordered of each product."""
    units: Counter[str] = Counter()
    for order in orders:
        units[order.product] += order.qty
    return units


def products_by_customer(orders: Iterable[Order]) -> dict[str, set[str]]:
    """Map each customer to the set of products they ordered."""
    products: defaultdict[str, set[str]] = defaultdict(set)
    for order in orders:
        products[order.customer].add(order.product)
    return dict(products)


def last_orders(orders: Iterable[Order], n: int) -> list[Order]:
    """Return the last n orders, reading the orders only once."""
    return list(orders)[-n:]


if __name__ == "__main__":
    print(units_by_product(read_orders("orders.csv")).most_common(2))
    print(sorted(products_by_customer(read_orders("orders.csv"))["ada"]))
    for order in last_orders(read_orders("orders.csv"), 2):
        print(order.customer, order.product, order.qty)

test_main.py

from learnrun import run_main

from main import Order, last_orders, read_orders


def test_last_two():
    """Die letzten zwei Bestellungen aus orders.csv"""
    got = last_orders(read_orders("orders.csv"), 2)
    assert got == [Order("bo", "tea, green", 2), Order("ada", "cake", 1)], f"last_orders(..., 2) lieferte {got!r}"


def test_zero():
    """n = 0 ergibt eine leere Liste"""
    got = last_orders(read_orders("orders.csv"), 0)
    assert got == [], f"last_orders(..., 0) lieferte {got!r}, erwartet: []"


def test_long_generator():
    """Ein langer Generator funktioniert"""
    got = [order.qty for order in last_orders((Order("c", "p", n) for n in range(10000)), 3)]
    assert got == [9997, 9998, 9999], f"die letzten drei Mengen sind {got!r}, erwartet: [9997, 9998, 9999]"


def test_report():
    """main.py gibt den Bericht aus"""
    got = run_main().strip().splitlines()
    expected = ["[('coffee', 4), ('tea', 3)]", "['cake', 'coffee', 'tea']", "bo tea, green 2", "ada cake 1"]
    assert got == expected, f"main.py gab {got!r} aus, erwartet: {expected!r}"

orders.csv

customer,product,qty
ada,tea,2
bo,coffee,1
ada,coffee,3
cy,tea,1
bo,"tea, green",2
ada,cake,1

Unter macOS und Linux tippen Sie python3, wo in diesen Befehlen python steht, wie in der ersten Lektion.

Programm ausführen:

python main.py

Prüfungen ausführen (learnrun.py muss im selben Ordner liegen):

python learnrun.py test

learnrun.py herunterladen

Häufige Fehler

Einen reader aus einem with-Block zurückgeben

import csv

with open("data.csv", "w", encoding="utf-8", newline="") as f:
    f.write("name,qty\ntea,2\n")


def read(path):
    with open(path, encoding="utf-8", newline="") as f:
        return csv.reader(f)


rows = read("data.csv")
print(next(rows))

Was Python ausgibt

ValueError: I/O operation on closed file.

Warum, und die Lösung

csv.reader ist lazy: Er liest die Datei erst, wenn eine Zeile verlangt wird. return verlässt den with-Block, der die Datei schließt, also findet das erste next() sie geschlossen vor. Machen Sie read stattdessen zum Generator: for row in csv.reader(f): yield row hält die Datei offen, bis die Zeilen verbraucht sind.

Ein Feld eines namedtuple ändern

from collections import namedtuple

Order = namedtuple("Order", ["product", "qty"])
order = Order("tea", 2)
order.qty = 3

Was Python ausgibt

AttributeError: can't set attribute

Warum, und die Lösung

Ein namedtuple ist ein Tupel, und Tupel sind unveränderlich. Erzeugen Sie stattdessen eine geänderte Kopie: order = order._replace(qty=3). Wenn sich Datensätze an Ort und Stelle ändern sollen, nehmen Sie eine Dataclass aus Lektion I2.

Mit einem normalen Dict zählen

counts = {}
for word in ["tea", "cake", "tea"]:
    counts[word] += 1
print(counts)

Was Python ausgibt

KeyError: 'tea'

Warum, und die Lösung

counts[word] += 1 liest zuerst counts[word], das es für ein neues Wort nicht gibt. Nutzen Sie counts = Counter() aus collections, das jeden Schlüssel bei 0 beginnt, oder zählen Sie in einem Schritt: Counter(["tea", "cake", "tea"]).

Python im Browser: Pyodide 314.0.7, MPL-2.0. Lizenz und Quellcode

Abschlussquiz

5 Fragen, ohne Hinweise. Ab 80 % ist die Lektion abgeschlossen.

Erledigen Sie zuerst alle Aufgaben oben, um das Abschlussquiz freizuschalten.

Kernideen

Counter und defaultdict behandeln fehlende Schlüssel

Zählen mit einem normalen Dict scheitert am ersten neuen Schlüssel: counts[word] += 1 löst KeyError aus. Ein Counter ist ein Dict, das für einen fehlenden Schlüssel 0 liefert, also funktioniert counts[word] += n einfach. Counter(iterable) zählt in einem Schritt, und most_common(n) liefert die n größten als Paare (Element, Anzahl). Ein defaultdict(factory) ruft für einen fehlenden Schlüssel factory() auf und speichert das Ergebnis: defaultdict(list) beginnt jeden Schlüssel mit [], defaultdict(set) mit einer leeren Menge. Wandeln Sie es vor der Rückgabe mit dict(...) um, damit spätere Tippfehler wieder KeyError auslösen.

namedtuple und deque

namedtuple("Order", ["customer", "product", "qty"]) erzeugt eine kleine Tupel-Klasse. order.qty liest sich besser als order[2], beides funktioniert, und die Ausgabe zeigt die Feldnamen. Wie jedes Tupel ist es unveränderlich: order._replace(qty=3) liefert eine geänderte Kopie. Eine deque ist eine listenähnliche Warteschlange, die an beiden Enden schnell hinzufügt und entfernt: popleft() ersetzt das langsame list.pop(0). Mit maxlen=n behält sie nur die letzten n Elemente und verwirft die ältesten. deque(iterable, maxlen=n) ist also das Ende jedes Iterables, auch eines, das nicht in den Speicher passt.

Eine lazy CSV-Pipeline

csv.reader zerlegt jede Zeile in eine Liste von Strings und behandelt Felder in Anführungszeichen wie "tea, green", an denen split(",") scheitert. csv.DictReader nutzt die Kopfzeile als Schlüssel, row["qty"] findet die Spalte also über ihren Namen. Öffnen Sie die Datei mit newline="", wie die csv-Doku verlangt. Werte bleiben Strings: Wandeln Sie sie selbst um. Legen Sie das Lesen in einen Generator, der pro Zeile einen Datensatz liefert, dann ziehen Counter oder deque die Zeilen einzeln durch. Die Datei liegt nie ganz im Speicher, und der with-Block schließt sie, wenn der Generator endet.

Quellen

Zuletzt geprüft am 29. September 2026

Projekt: ein lazy CSV-Bericht mit collections

Aufwärmen aus Lektion I3.2: Einen Generator können Sie einmal durchlaufen. Was gibt das aus?

Sagen Sie voraus, bevor Sie weiterlesen: z kommt in "banana" nicht vor. Was gibt das aus?

Ergänzen Sie die Factory, damit jeder neue Anfangsbuchstabe mit einer leeren Liste beginnt.

Fünf Zahlen kommen in eine deque, die drei fasst. Was gibt das aus?

Counter, defaultdict, deque und namedtuple sind aus collections importiert. Ordnen Sie jedem Ausdruck seinen Wert zu.

Knobelaufgabe. Die deque behält nur ein Element. Was gibt das aus?

Selbst programmieren

Seitenaufrufe, gezählt und gruppiert

Übungen

Schritt 1: Bestellungen lazy lesen

Schritt 2: Mengen mit Counter zählen

Schritt 3: Produkte mit defaultdict gruppieren

Schritt 4: die neuesten Bestellungen und der Bericht

Häufige Fehler

Einen reader aus einem with-Block zurückgeben

Ein Feld eines namedtuple ändern

Mit einem normalen Dict zählen

Abschlussquiz

Kernideen

Counter und defaultdict behandeln fehlende Schlüssel

namedtuple und deque

Eine lazy CSV-Pipeline

Quellen