Claude Code & asyncio: Asynchrones Python meistern

Asynchrones Python ist eine der leistungsstärksten Techniken in der modernen Backend-Entwicklung — und gleichzeitig eine der am häufigsten falsch verstandenen. asyncio, async/await, Event-Loop, Coroutines: Die Konzepte klingen komplex, und der erste Fehler mit einem blockierenden Call mitten im asynchronen Code kann eine ganze Stunde kosten, bevor man versteht, was eigentlich passiert.

Claude Code verändert diese Lernkurve grundlegend. Nicht weil es asyncio einfacher macht — das Modell ist, was es ist — sondern weil es im richtigen Moment erklärt, auf den eigenen Code angewendet, und beim Debugging tatsächlich mitdenkt statt nur Dokumentationsauszüge zu liefern.

1. Was ist asyncio — und warum braucht man es?

Python ist von Haus aus single-threaded. Ein normaler Python-Prozess kann immer nur eine Sache gleichzeitig tun. Das ist kein Problem, solange man nur CPU-intensive Aufgaben hat — Berechnungen, Transformationen, Datenverarbeitung. Das Problem entsteht bei I/O-Operationen: HTTP-Requests, Datenbankabfragen, Datei-Lesen. Diese sind langsam, und während Python auf die Antwort wartet, tut es buchstäblich nichts.

asyncio löst dieses Problem mit einem Event-Loop: einem zentralen Koordinator, der verwaltet, welche Coroutine gerade läuft und welche auf I/O wartet. Wenn Coroutine A auf einen HTTP-Response wartet, führt der Event-Loop Coroutine B aus. Sobald die Antwort für A eintrifft, wird A fortgesetzt. So entsteht echter Parallelismus im I/O-Bereich — ohne Threads, ohne Locks, ohne Race Conditions auf geteiltem State.

2. async/await: die Grundsyntax

Eine Coroutine ist eine Funktion, die mit async def definiert ist. Sie kann an bestimmten Punkten pausieren — markiert mit await — und dem Event-Loop die Kontrolle zurückgeben, während sie auf das Ergebnis wartet.

import asyncio

async def hole_daten(url: str) -> str:
    # Simuliert eine I/O-Operation
    await asyncio.sleep(1)
    return f"Daten von {url}"

async def main():
    ergebnis = await hole_daten("https://api.example.com/data")
    print(ergebnis)

asyncio.run(main())

Das await vor asyncio.sleep(1) ist entscheidend: es sagt dem Event-Loop "hier kann ich kurz pausieren, mach ruhig was anderes". Ohne await läuft die Funktion synchron durch und blockiert den gesamten Event-Loop.

Claude Code hilft besonders beim Einstieg: zeige ihm synchronen Python-Code und frage explizit nach der asynchronen Version. Es liest die Funktion, erkennt die I/O-Stellen und schlägt die korrekte Umstrukturierung vor — inklusive der Imports, die man vergisst, und der Fehler, die man erst nach dem ersten Ausführen sieht.

3. asyncio.gather() — parallele Ausführung

Der eigentliche Gewinn von asyncio entsteht, wenn mehrere Coroutinen gleichzeitig warten. asyncio.gather() startet mehrere Coroutinen parallel und wartet, bis alle fertig sind:

import asyncio

async def hole_nutzer(nutzer_id: int) -> dict:
    await asyncio.sleep(0.5)  # Datenbankabfrage simuliert
    return {"id": nutzer_id, "name": f"Nutzer {nutzer_id}"}

async def main():
    # Alle drei Abfragen laufen gleichzeitig
    ergebnisse = await asyncio.gather(
        hole_nutzer(1),
        hole_nutzer(2),
        hole_nutzer(3),
    )
    print(ergebnisse)
    # Laufzeit: ~0.5s statt 1.5s

asyncio.run(main())

Ohne gather() würden die drei Abfragen sequenziell laufen: 0.5 + 0.5 + 0.5 = 1.5 Sekunden. Mit gather() laufen sie parallel: ~0.5 Sekunden. Bei realen API-Calls mit 200ms Latenz macht das den Unterschied zwischen einer trägen und einer reaktiven Anwendung.

Fehlerbehandlung in gather()

Standardmäßig bricht gather() beim ersten Fehler ab und hebt die Exception weiter. Mit return_exceptions=True werden Fehler als Ergebnisse zurückgegeben, sodass alle Coroutinen abgeschlossen werden:

ergebnisse = await asyncio.gather(
    hole_nutzer(1),
    hole_nutzer(999),  # Dieser schlägt fehl
    hole_nutzer(3),
    return_exceptions=True
)
# [{"id": 1, ...}, UserNotFoundError(999), {"id": 3, ...}]

4. asyncio.create_task() — Tasks im Hintergrund

Während gather() wartet, bis alle Coroutinen fertig sind, ermöglicht create_task() das Starten einer Coroutine im Hintergrund — ohne sofort zu warten:

async def main():
    # Task startet sofort, wir warten nicht
    task = asyncio.create_task(hole_daten("https://api.example.com"))

    # Andere Arbeit erledigen
    await verarbeite_lokale_daten()

    # Jetzt auf den Task warten
    ergebnis = await task
    print(ergebnis)

Tasks eignen sich besonders für Hintergrundoperationen: Logging, Cache-Aktualisierungen, Benachrichtigungen. Sie laufen unabhängig und blockieren den Hauptfluss nicht.

5. Timeouts mit asyncio.wait_for()

Externe APIs können langsam oder nicht erreichbar sein. Ohne Timeout hängt dein Code ewig. asyncio.wait_for() gibt einer Coroutine ein Zeitlimit:

import asyncio

async def langsame_api():
    await asyncio.sleep(10)  # Simuliert eine langsame API
    return "Daten"

async def main():
    try:
        ergebnis = await asyncio.wait_for(
            langsame_api(),
            timeout=3.0  # Maximal 3 Sekunden
        )
    except asyncio.TimeoutError:
        print("API hat nicht rechtzeitig geantwortet")
        ergebnis = None

Claude Code hilft beim richtigen Umgang mit TimeoutError: es zeigt nicht nur den try/except-Block, sondern analysiert den Code-Kontext und empfiehlt, ob ein Retry-Mechanismus, ein Fallback-Wert oder eine explizite Fehlermeldung sinnvoller ist.

6. asyncio.Queue — Producer-Consumer-Muster

asyncio.Queue implementiert eine thread-sichere Queue für asynchrone Produzenten und Konsumenten — ideal für Pipelines, bei denen Daten erzeugt und verarbeitet werden:

import asyncio

async def produzent(queue: asyncio.Queue):
    for i in range(10):
        await queue.put(f"Aufgabe {i}")
        await asyncio.sleep(0.1)
    await queue.put(None)  # Sentinel: Produktion beendet

async def konsument(queue: asyncio.Queue):
    while True:
        aufgabe = await queue.get()
        if aufgabe is None:
            break
        print(f"Verarbeite: {aufgabe}")
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=5)
    await asyncio.gather(
        produzent(queue),
        konsument(queue),
    )

Das maxsize-Argument verhindert, dass der Produzent zu weit voraus läuft: sobald die Queue voll ist, wartet queue.put() automatisch. Das macht Backpressure ohne explizite Synchronisation möglich.

7. aiohttp — HTTP-Requests asynchron

Das Standard-requests-Paket ist synchron: jeder HTTP-Call blockiert den Event-Loop. aiohttp ist der asynchrone Ersatz:

import asyncio
import aiohttp

async def hole_json(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as response:
        response.raise_for_status()
        return await response.json()

async def main():
    urls = [
        "https://api.example.com/nutzer/1",
        "https://api.example.com/nutzer/2",
        "https://api.example.com/nutzer/3",
    ]
    async with aiohttp.ClientSession() as session:
        ergebnisse = await asyncio.gather(
            *[hole_json(session, url) for url in urls]
        )
    print(ergebnisse)

Der async with-Block stellt sicher, dass die Session korrekt geschlossen wird — auch bei Fehlern. Eine gemeinsame Session für alle Requests zu verwenden ist dabei deutlich effizienter als eine neue Session pro Request zu öffnen.

8. Die häufigsten Fallen — blocking calls

Der gefährlichste Fehler in asyncio-Code: eine synchrone, blockierende Funktion in einer Coroutine aufrufen. Sie friert den gesamten Event-Loop ein — kein anderer Task läuft, während sie wartet.

# FALSCH: blockiert den Event-Loop
async def lade_datei(pfad: str):
    with open(pfad) as f:  # synchron!
        return f.read()

# RICHTIG: mit asyncio-kompatiblem I/O
import aiofiles

async def lade_datei(pfad: str):
    async with aiofiles.open(pfad) as f:
        return await f.read()

Weitere typische Fallen:

• time.sleep() statt await asyncio.sleep() — blockiert komplett
• Synchrone Datenbankbibliotheken wie psycopg2 statt asyncpg
• CPU-intensive Operationen direkt in Coroutinen — stattdessen loop.run_in_executor() verwenden
• Verschachtelte asyncio.run()-Aufrufe — das ist verboten; nur auf der obersten Ebene erlaubt

"Die Funktion requests.get() in Zeile 23 blockiert deinen Event-Loop. Ersetze sie durch aiohttp.ClientSession.get() und mache die Funktion async. Die Session solltest du als Parameter übergeben oder in einem Context-Manager auf Applikationsebene verwalten."

Genau diese Art Feedback bekommst du von Claude Code, wenn du es nach einem Performance-Problem fragst. Nicht "verwende aiohttp statt requests" — sondern warum, wo, und wie konkret in deinem Code.

9. Claude Code-Tipps für asyncio

Einige Prompts, die sich in der Praxis bewährt haben:

• Synchronen Code migrieren: claude "Mache diese Funktion asynchron und erkläre jeden Schritt"
• Performance analysieren: claude "Warum ist dieser asyncio-Code langsam? Wo blockiere ich den Event-Loop?"
• Muster erklären lassen: claude "Zeige mir den Unterschied zwischen gather() und create_task() an einem konkreten Beispiel"
• Tests schreiben: claude "Schreibe pytest-asyncio-Tests für diese Coroutine"

Claude Code liest dabei immer den tatsächlichen Code im Verzeichnis — nicht ein abstraktes Beispiel. Es sieht die konkrete Implementierung, die Imports, die Abhängigkeiten. Die Empfehlungen passen dadurch zum realen Kontext, nicht zu einem Lehrbuchbeispiel.