Python asyncio: Async/Await in der Praxis

Asynchrone Programmierung mit asyncio ist seit Python 3.5 ein fester Bestandteil der Sprache. Django unterstützt seit Version 3.1 Async-Views. Zeit, das Thema richtig zu durchdringen.

Das Konzept: Event Loop und Coroutines

Normaler Python-Code läuft synchron: eine Operation blockiert den Thread bis sie fertig ist. asyncio nutzt eine Event Loop: Während eine Operation wartet (z.B. auf eine HTTP-Antwort), kann die Event Loop eine andere Coroutine fortsetzen.

asyncio ist ideal für I/O-bound Operationen (API-Aufrufe, Datenbankabfragen, Datei-I/O). Für CPU-bound Aufgaben ist multiprocessing besser.

Erste Schritte mit async/await

import asyncio
import httpx  # async-fähiger HTTP-Client

async def fetch_user(user_id: int) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/users/{user_id}")
        return response.json()

async def main():
    user = await fetch_user(42)
    print(user["name"])

# Entry point
asyncio.run(main())

Tasks und gather() für parallele Ausführung

Der eigentliche Vorteil von asyncio: mehrere Operationen parallel starten:

import asyncio
import httpx

async def fetch_user(client, user_id):
    r = await client.get(f"/users/{user_id}")
    return r.json()

async def fetch_all_users(user_ids: list) -> list:
    async with httpx.AsyncClient(base_url="https://api.example.com") as client:
        # Alle Requests gleichzeitig starten (nicht nacheinander!)
        tasks = [fetch_user(client, uid) for uid in user_ids]
        results = await asyncio.gather(*tasks)
    return results

# 100 Users parallel laden statt sequenziell
users = asyncio.run(fetch_all_users(range(1, 101)))

# gather() mit Fehlerbehandlung
results = await asyncio.gather(
    fetch_user(client, 1),
    fetch_user(client, 2),
    fetch_user(client, 999),  # Existiert nicht
    return_exceptions=True   # Fehler nicht werfen, als Ergebnis zurückgeben
)

for r in results:
    if isinstance(r, Exception):
        print(f"Fehler: {r}")
    else:
        print(r["name"])

Async in Django

Seit Django 3.1 können Views, Middleware und Tests async sein:

# Async View
from django.http import JsonResponse
import httpx

async def external_data_view(request):
    async with httpx.AsyncClient() as client:
        r1, r2 = await asyncio.gather(
            client.get("https://api1.example.com/data"),
            client.get("https://api2.example.com/data"),
        )
    return JsonResponse({"api1": r1.json(), "api2": r2.json()})

# ORM in Async-Views: sync_to_async verwenden
from asgiref.sync import sync_to_async

async def user_list(request):
    # ORM-Calls sind synchron - in Thread-Pool ausführen
    users = await sync_to_async(list)(
        User.objects.filter(active=True).values("id", "email")
    )
    return JsonResponse({"users": users})

# Oder mit dem async ORM (Django 4.1+)
async def user_list_v2(request):
    users = [u async for u in User.objects.filter(active=True).values("id", "email")]
    return JsonResponse({"users": users})

Häufige Fallstricke

• Keine blockierenden Aufrufe in Coroutines: requests.get(), time.sleep() blockieren die Event Loop. Nutze httpx/aiohttp und asyncio.sleep()
• asyncio ist kein Threading: Nur eine Coroutine läuft gleichzeitig. CPU-intensive Arbeit blockiert trotzdem
• asyncio.run() nur einmal: Nicht in einer bereits laufenden Event Loop aufrufen
• Django ORM ist synchron: Immer sync_to_async oder async ORM (4.1+) verwenden

Yevhen Chubchyk

Senior Python / Django Entwickler · Freelancer seit 2016 · 20+ Jahre IT-Erfahrung. Kunden: Siemens Healthineers, Engie, natureOffice u.a.