Context Manager und with-Anweisung in Python

1. Übersicht

Datei öffnen, Datenbankverbindung aufbauen oder einen Lock halten – das erfordert jeweils einen Aufräumschritt am Ende. Dieser Schritt muss laufen, auch wenn der Code dazwischen eine Exception wirft. Python löst das mit Context Managern und der with-Anweisung.

Dieser Artikel zeigt, wie with intern funktioniert und wie du eigene Context Manager mit Klassen oder mit contextlib schreibst. Du wirst auch sehen, wie man Exception in __exit__ behandelt, wie man Context Manager testet und welche Fehler häufig vorkommen.

2. Problem: Ressourcen-Cleanup mit try/finally

Hier ist das traditionelle Vorgehen, um eine Datei zu lesen, ohne with zu verwenden:

f = open("orders.csv", encoding="utf-8")
try:
    header = f.readline()
    print(header)
finally:
    f.close()

Das ist korrekt, aber ausschweifend. Es lässt sich auch leicht vergessen. Wenn du open() in den try-Block stellst und open() fehlschlägt, wird f nie zugewiesen und der finally-Block wirft einen NameError. Die with-Anweisung komprimiert dieses ganze Muster:

with open("orders.csv", encoding="utf-8") as f:
    header = f.readline()
    print(header)

print(f.closed)

order_id,customer,amount

True

Wenn der with-Block verlassen wird, ist die Datei geschlossen. Das gilt sowohl für einen normalen Ausstieg als auch für eine Exception.

3. Das Context-Manager-Protokoll

Jedes Objekt mit den Methoden __enter__ und __exit__ kann mit with verwendet werden. Konzeptionell ist die Anweisung:

with EXPR as target:
    BODY

äquivalent zu:

manager = EXPR
target = manager.__enter__()
try:
    BODY
except BaseException as e:
    if not manager.__exit__(type(e), e, e.__traceback__):
        raise
else:
    manager.__exit__(None, None, None)

Drei Punkte sind wichtig:

• Der Wert, der target zugewiesen wird, ist das Ergebnis von __enter__(), nicht EXPR selbst.
• __exit__ wird immer aufgerufen. Bei einer Exception erhält sie (exc_type, exc_value, traceback). Ohne Fehler sind alle drei Parameter None.
• Wenn __exit__ einen Wert truthy zurückgibt, wird die Exception unterdrückt. Sonst wird sie weitergegeben.

3.1. Context Manager mit Klassen schreiben

Das erste Beispiel ist ein Timer, der die Ausführungszeit eines Code-Blocks misst:

import time


class Timer:
    def __init__(self, label: str):
        self.label = label
        self.elapsed: float | None = None

    def __enter__(self):
        self._start = time.perf_counter()
        return self

    def __exit__(self, exc_type, exc, tb):
        self.elapsed = time.perf_counter() - self._start
        print(f"[{self.label}] {self.elapsed:.3f}s")
        return False


with Timer("summe_berechnen") as t:
    total = sum(range(5_000_000))

print(t.elapsed > 0)

[summe_berechnen] 0.081s
True

Erklärung Teil für Teil:

• __enter__ speichert die Startzeit und gibt self zurück. Dadurch ist t das Timer-Objekt selbst.
• __exit__ berechnet die verstrichene Zeit, gibt sie aus und gibt False zurück. Eine Exception, falls vorhanden, wird nicht unterdrückt.
• Die Variable t kann nach dem with-Block noch verwendet werden. Der Grund: with erzeugt keinen neuen Scope.

3.2. Exception in __exit__ behandeln

Context Manager zeigen ihre Stärke am deutlichsten, wenn du unterschiedlich reagieren möchtest, je nachdem ob der Block erfolgreich war oder fehlgeschlagen ist. Das folgende Beispiel ist eine einfache Transaction in einem Ledger:

class Transaction:
    def __init__(self, ledger: list):
        self.ledger = ledger
        self.pending: list = []

    def __enter__(self):
        return self.pending

    def __exit__(self, exc_type, exc, tb):
        if exc_type is None:
            self.ledger.extend(self.pending)
            print(f"{len(self.pending)} Einträge committed")
        else:
            print(f"rollback wegen {exc_type.__name__}: {exc}")
        return False


ledger = []

with Transaction(ledger) as tx:
    tx.append({"order_id": 1, "amount": 250_000})
    tx.append({"order_id": 2, "amount": 120_000})

try:
    with Transaction(ledger) as tx:
        tx.append({"order_id": 3, "amount": 90_000})
        raise ValueError("Guthaben reicht nicht aus")
except ValueError:
    pass

print(len(ledger))

2 Einträge committed
rollback wegen ValueError: Guthaben reicht nicht aus
2

Im zweiten with-Block wird Bestellung 3 nicht ins ledger geschrieben. Da __exit__ False zurückgibt, wird ValueError immer noch weitergegeben und der Aufrufer kann entscheiden, wie er damit umgeht.

Warnung: Gib nur True von __exit__ zurück, wenn du bewusst die Exception unterdrücken möchtest. Eine versehentliche Rückgabe von self oder einem anderen truthy-Objekt macht Fehler spurlos verschwinden.

4. Kürzer mit @contextmanager

Eine ganze Klasse nur für Setup und Teardown zu schreiben ist ziemlich aufwändig. Das Modul contextlib stellt den Decorator @contextmanager zur Verfügung, um einen Generator in einen Context Manager umzuwandeln:

from contextlib import contextmanager


class PaymentService:
    def connect(self):
        print("Zahlungs-Gateway verbunden")

    def charge(self, customer: str, amount: int):
        print(f"Berechne {amount} für {customer}")

    def disconnect(self):
        print("Verbindung getrennt")


@contextmanager
def payment_session(service: PaymentService):
    service.connect()
    try:
        yield service
    finally:
        service.disconnect()


with payment_session(PaymentService()) as payment:
    payment.charge("Lan", 300_000)

Zahlungs-Gateway verbunden
Berechne 300000 für Lan
Verbindung getrennt

Wie der Decorator den Generator zum Context-Manager-Protokoll abbildet:

• Code vor yield entspricht __enter__.
• Der Wert, der mit yield zurückgegeben wird, ist der Wert nach as.
• Code nach yield entspricht __exit__. Wenn der with-Block eine Exception wirft, wird diese an der Stelle von yield weitergegeben.

Da die Exception direkt bei yield erscheint, ist ein try/finally um yield erforderlich. Ohne ihn wird disconnect() bei Fehlern nicht ausgeführt.

Um verschiedene Fehlertypen unterschiedlich zu behandeln, kannst du zusätzlich except verwenden:

@contextmanager
def safe_charge(order_id: int):
    try:
        yield
    except ConnectionError as e:
        print(f"Bestellung {order_id}: Netzwerkfehler, wird später wiederholt ({e})")


with safe_charge(42):
    raise ConnectionError("timeout")

print("Verarbeitung läuft weiter")

Bestellung 42: Netzwerkfehler, wird später wiederholt (timeout)
Verarbeitung läuft weiter

Der Generator fängt den Fehler ab, wirft ihn aber nicht weiter, also wird die Exception unterdrückt. Das entspricht der __exit__-Methode, die True zurückgibt.

5. Praktische Tools aus contextlib

5.1. suppress: Eine bestimmte Exception ignorieren

import os
from contextlib import suppress

with suppress(FileNotFoundError):
    os.remove("cache_orders.tmp")

print("fertig")

fertig

Das ist kürzer und aussagekräftiger als try: ... except FileNotFoundError: pass.

5.2. ExitStack: Ressourcen mit unbekannter Anzahl

Angenommen, du möchtest mehrere Rechnungsdateien zusammenführen, aber die Anzahl ist erst zur Laufzeit bekannt:

from contextlib import ExitStack
from pathlib import Path


def merge_invoices(paths: list[Path], output: Path) -> None:
    with ExitStack() as stack:
        files = [stack.enter_context(open(p, encoding="utf-8")) for p in paths]
        with open(output, "w", encoding="utf-8") as out:
            for f in files:
                out.write(f.read())


for i in range(3):
    Path(f"rechnung_{i}.txt").write_text(f"Rechnung {i}\n", encoding="utf-8")

merge_invoices([Path(f"rechnung_{i}.txt") for i in range(3)], Path("all.txt"))
print(Path("all.txt").read_text(encoding="utf-8"))

Rechnung 0
Rechnung 1
Rechnung 2

enter_context ruft __enter__ auf und merkt sich das entsprechende __exit__. Wenn ExitStack beendet wird, werden die Dateien in umgekehrter Reihenfolge geschlossen. Falls die dritte Datei nicht geöffnet werden kann, werden die ersten beiden trotzdem korrekt geschlossen.

5.3. closing und nullcontext

• closing(obj) wandelt ein Objekt mit nur einer close()-Methode in einen Context Manager um.
• nullcontext(value) ist ein Context Manager, der nichts tut. Das ist nützlich, wenn ein Parameter ein Context Manager sein kann oder nicht:

from contextlib import nullcontext
import threading

def update_stock(product: dict, qty: int, lock: threading.Lock | None = None):
    with lock if lock is not None else nullcontext():
        product["stock"] -= qty

product = {"sku": "AO-01", "stock": 10}
update_stock(product, 3)
update_stock(product, 2, threading.Lock())
print(product["stock"])

5

6. Mehrere Context Manager in einer with-Anweisung

Du kannst mehrere Context Manager in einer einzigen with-Anweisung deklarieren. Ab Python 3.10 ermöglicht die Syntax mit Klammern ein sauberes Zeilenumbruch:

with (
    open("orders.csv", encoding="utf-8") as src,
    open("orders_backup.csv", "w", encoding="utf-8") as dst,
):
    dst.write(src.read())

Das ist gleichbedeutend mit zwei verschachtelten with-Blöcken. src wird zuerst geöffnet und zuletzt geschlossen.

7. Asynchrone Context Manager

Mit asyncio-Code ist das entsprechende Protokoll __aenter__ und __aexit__, das mit async with verwendet wird. Die Generator-Version davon ist @asynccontextmanager:

import asyncio
from contextlib import asynccontextmanager


@asynccontextmanager
async def order_queue(name: str):
    print(f"Queue {name} verbunden")
    await asyncio.sleep(0.01)
    try:
        yield [f"order-{i}" for i in range(3)]
    finally:
        await asyncio.sleep(0.01)
        print(f"Queue {name} geschlossen")


async def main():
    async with order_queue("payments") as messages:
        for m in messages:
            print("verarbeite", m)

asyncio.run(main())

Queue payments verbunden
verarbeite order-0
verarbeite order-1
verarbeite order-2
Queue payments geschlossen

Warnung: Die Verwendung von with (statt async with) mit einem asynchronen Context Manager führt zu TypeError (oder AttributeError in älteren Python-Versionen). Umgekehrt genauso. Die beiden Protokolle sind getrennt.

8. Context Manager testen

Context Manager sollten sowohl den erfolgreichen Pfad als auch den Fehlerpfad getestet werden. Installiere zuerst pytest:

pip install pytest

Speichere die Klasse Transaction aus Abschnitt 3.2 in transaction.py, dann schreibe die Tests:

# test_transaction.py
import pytest
from transaction import Transaction


def test_commit_when_no_error():
    ledger = []
    with Transaction(ledger) as tx:
        tx.append({"order_id": 1, "amount": 100})

    assert ledger == [{"order_id": 1, "amount": 100}]


def test_rollback_and_propagate_error():
    ledger = []
    with pytest.raises(ValueError, match="Guthaben"):
        with Transaction(ledger) as tx:
            tx.append({"order_id": 2, "amount": 999})
            raise ValueError("Guthaben reicht nicht aus")

    assert ledger == []

pytest -q

..                                                       [100%]
2 passed in 0.01s

Der zweite Test prüft zwei Dinge gleichzeitig: Die Daten werden zurückgerollt, und die Exception wird nicht unterdrückt. pytest.raises selbst ist auch ein Context Manager – ein bekanntes Beispiel genau dieses Themas.

9. Häufige Fehler und wann man es nicht verwenden sollte

try/finally um yield vergessen. Das ist der häufigste Fehler bei @contextmanager:

@contextmanager
def broken_session(service):
    service.connect()
    yield service
    service.disconnect()  # läuft nicht, wenn with eine Exception wirft

Ressource nach dem with verwenden. Die Variable existiert noch, aber die Ressource ist freigegeben:

with open("orders.csv", encoding="utf-8") as f:
    pass
f.read()

ValueError: I/O operation on closed file.

Mehrfaches yield im Generator. Ein Generator mit @contextmanager darf genau einmal yield. Bei mehrfachen Yields wirft Python RuntimeError: generator didn't stop.

Context Manager von @contextmanager wiederverwenden. Dieses Objekt kann nur einmal verwendet werden. Ein zweites with auf demselben Objekt schlägt fehl. Jedes Mal, wenn du es brauchst, rufe die Funktion erneut auf, um ein neues Objekt zu erzeugen.

Übergebrauch. Context Manager sollten nicht in diesen Fällen verwendet werden:

• Ressourcen müssen länger als ein Code-Block leben, z.B. ein Connection Pool über die gesamte Lebensdauer der Anwendung. Verwalte sie dann im Startup und Shutdown.
• Die Logik hat kein klares “öffnen/schließen” Paar. Eine gewöhnliche Funktion ist dann lesbarer.
• Du möchtest alle Exceptions mit suppress(Exception) ignorieren. Das versteckt Bugs genauso wie except: pass.

Warnung: Verlasse dich nicht auf Garbage Collection oder __del__ zum Schließen von Ressourcen. Wann diese ausgeführt werden, ist nicht garantiert, besonders auf Implementierungen wie PyPy. Verwende with.

10. Fazit

Context Manager kapseln Setup und Teardown an einer Stelle ein und garantieren, dass Teardown auch bei Exceptions läuft. Du hast gesehen:

• Das __enter__ und __exit__ Protokoll, und was der Rückgabewert von __exit__ bedeutet.
• Wie man mit @contextmanager kürzer schreibt, mit try/finally um yield als Erfordernis.
• Praktische Tools wie suppress, ExitStack, closing, nullcontext, und die asynchrone Variante.
• Wie man Context Manager mit pytest testet – sowohl Commit- als auch Rollback-Pfad.

Wenn dein Code ein “nimm-dann-gib-zurück” Paar hat (Datei, Lock, Transaction, Verbindung), solltest du überlegen, es in einen Context Manager zu packen.