Die Python-Funktion anext() ermöglicht das sichere und kontrollierte Abrufen des nächsten Elements aus einem Iterator. Sie erweitert die Funktionalität von next() um die Möglichkeit, einen Standardwert zu definieren, der zurückgegeben wird, falls der Iterator erschöpft ist. Dies erleichtert die Fehlerbehandlung und sorgt für robusteren Code beim Arbeiten mit Iteratoren in Python.

01 · Abschnitt

Einführung

Die Funktion anext() ist eine Builtin-Funktion in Python, die erst in Version 3.10 eingeführt wurde. Sie stellt das asynchrone Äquivalent zur bekannten next() Funktion dar und ist ein Baustein für Arbeit mit asynchronen Iteratoren.

Die Funktion anext() ist eine Coroutine-Funktion, die das nächste Element aus einem asynchronen Iterator abruft. Sie löst ein wichtiges Problem: Das sequentielle Durchlaufen von Datenstrukturen, die asynchron generiert werden.

Welches Problem wird gelöst?

Vor der Einführung von anext() mussten Entwickler komplexere Konstrukte verwenden, um einzelne Elemente aus asynchronen Iteratoren zu extrahieren.

Die Funktion vereinfacht:

  • Asynchrone Iteration: Einfacher Zugriff auf das nächste Element eines async Iterators
  • Fehlerbehandlung: Elegante Behandlung von StopAsyncIteration Ausnahme
  • Lesbarkeit: Reduziert Boilerplate-Code bei asynchronen Operationen
  • Konsequenz: Parallele API zur synchronen next() Funktion
02 · Abschnitt

Syntax

Python Syntax 
awaitable anext(async_iterator)
awaitable anext(async_iterator, default)
Parameter
async_iterator

Ein Objekt, das das async iterator Protokoll implementiert.

default

(Optional) Rückgabewert, wenn der Iterator erschöpft (verbraucht) ist. Verhindert das Auftreten einer StopAsyncIteration Exception. Standard nicht gesetzt.

03 · Abschnitt

Rückgabewert

Bei erfolgreichem Aufruf:

  • Typ: Awaitable/Coroutine
  • Inhalt: Das nächste Element aus dem asynchronen Iterator
  • Verhalten: Muss mit await aufgerufen werden

Bei erschöpften/verbrauchten Iterator:

  • Mit default: Gibt den default Wert zurück
  • Ohne default: Löst StopAsyncIteration Exception aus
04 · Abschnitt

Besonderheiten

Coroutine-Verhalten

anext() ist eine Coroutine-Funktion und muss daher immer mit await aufgerufen werden.

Python 
# Korrekt
result = await anext(async_iter)

# Inkorrekt - gibt Coroutine-Objekt zurück
result = anext(async_iter)

Exception Handling

Die Funktion kann verschiedene Exceptions auslösen:

  • StopAsyncIteration: Iterator ist erschöpft (ohne default Parameter)
  • TypeError: Parameter ist kein async iterator
05 · Abschnitt

Async Iterator Protokoll

Was macht ein Objekt zu einem Async Iterator? Ein Objekt ist ein Async Iterator, wenn es die Methoden __aiter__() und __anext__() implementiert.

Python 
class AsyncIterator:

    def __aiter__(self):
        return self

    def __anext__(self):
        # Logik zur Rückgabe des nächsten
        # Elements
        pass

Async Generatoren implementieren automatisch das Async Iterator Protokoll.

Python 
async def async_gen():
    for i in range(3):
        yield i

# async_gen() ist automatisch ein async iterator
06 · Abschnitt

Praktische Beispiele

Grundlegende Verwendung

Python Beispiel 
import asyncio

async def simple_async_generator():
    """
    Einfacher async generator.
    """
    for i in range(5):
        await asyncio.sleep(0.1)
        yield f"Item {i}"

async def basic_anext_example():
    # Erstelle async iterator
    async_iter = simple_async_generator()

    # Hole erstes Element
    first_item = await anext(async_iter)
    print(f"Erstes Element: {first_item}")

    # Hole zweites Element
    second_item = await anext(async_iter)
    print(f"Zweites Element: {second_item}")

    # Hole drittes Element
    third_item = await anext(async_iter)
    print(f"Drittes Element: {third_item}")

asyncio.run(basic_anext_example)
Output 
Erstes Element: Item 0
Zweites Element: Item 1
Drittes Element: Item 2

Verwendung mit Default-Wert

Python Beispiel - Default-Wert 
import asyncio

async def limited_generator():
    """Generator mit nur 2 Elementen"""
    yield "Erste Nachricht"
    yield "Zweite Nachricht"

async def anext_with_default_example():
    async_iter = limited_generator()

    # Normale Elemente abrufen
    msg_one = await anext(async_iter, "Standard")
    print(f"Nachricht 1: {msg_one}")

    msg_two = await anext(async_iter, "Standard")
    print(f"Nachricht 2: {msg_two}")

    # Iterator ist verbraucht - default wird zurückgegeben
    msg_three = await anext(async_iter, "Standard")
    print(f"Nachricht 3: {msg_three}")

    # Noch ein Versuch
    msg_four = await anext(async_iter, "Standard")
    print(f"Nachricht 4: {msg_four}")

asyncio.run(anext_with_default_example())
Output 
Nachricht 1: Erstes Element
Nachricht 2: Zweites Element
Nachricht 3: Standard
Nachricht 4: Standard

Exception Handling ohne Default

Python Beispiel - Exception Handling 
import asyncio

async def short_generator():
    """Generator mit nur 1 Element"""
    yield "Einziges Element"


async def anext_exception_handling():
    async_iter = short_generator()

    try:
        # Erstes Element - erfolgreich
        item_one = await anext(async_iter)
        print(f"Erhalten: {item_one}")

        # Zweites Element - Iterator ist verbraucht
        item_two = await anext(async_iter)
        print(f"Erhalten: {item_two}")

    except StopAsyncIteration:
        print("Iterator ist verbraucht - keine weiteren Elemente")


asyncio.run(anext_exception_handling())
Output 
Erhalten: Einziges Element
Iterator ist verbraucht - keine weiteren Elemente

Komplexe async Iterator Klasse

Python Beispiel - Iterator Klasse 
import random
import asyncio

class AsyncDataStream:
    """Simuliert einen async Datenstream"""

    def __init__(self, max_items=5):
        self.max_items = max_items
        self.current_count = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.current_count >= self.max_items:
            raise StopAsyncIteration

        # Simuliere Netzwerk-Latenz
        await asyncio.sleep(random.uniform(0.1, 0.5))

        self.current_count += 1

        return {
            "id": self.current_count,
            "data": f"Datenpaket #{self.current_count}",
            "timestamp": asyncio.get_event_loop().time()
        }


async def process_data_stream():
    """Verarbeitet Daten mit anext()"""
    stream = AsyncDataStream(max_items=3)

    print("Beginne Datenverarbeitung ...")

    # Verarbeite erste zwei Pakete einzeln
    for i in range(2):
        try:
            data_packet = await anext(stream)
            print(f"Verarbeitet: {data_packet}")
        except StopAsyncIteration:
            print("Stream unerwartet beendet")
            break

    # Versuche weitere Pakete mit Default
    while True:
        next_packet = await anext(stream, None)
        if next_packet is None:
            print("Stream beendet")
            break

        print(f"Zusätzliches Paket: {next_packet}")


asyncio.run(process_data_stream())
Output 
Beginne Datenverarbeitung ...
Verarbeitet: {'id': 1, 'data': 'Datenpaket #1', 'timestamp': 480499.533708145}
Verarbeitet: {'id': 2, 'data': 'Datenpaket #2', 'timestamp': 480499.955362481}
Zusätzliches Paket: {'id': 3, 'data': 'Datenpaket #3', 'timestamp': 480500.339291559}
Stream beendet

Async Queue Iterator

Python Beispiel - Async Queue Iterator 
import asyncio 

class AsyncQueueIterator:
    """
    Wandelt eine asyncio.Queue in einen
    async Iterator um.
    """

    def __init__(self, queue, sentinel=None):
        self.queue = queue
        self.sentinel = sentinel # Wert, der das Ende signalisiert

    def __aiter__(self):
        return self

    async def __anext__(self):
        item = await self.queue.get()

        if item == self.sentinel:
            raise StopAsyncIteration
        
        return item

async def producer(queue):
    """Produziert Elemente für die Queue"""
    for i in range(5):
        await asyncio.sleep(0.2)
        await queue.put(f"Produziert: Item {i}")

    # Signalisiere Ende
    await queue.put(None) # Sentinel-Wert

async def consumer_with_anext():
    """Konsumiert Elemente mit anext()"""
    queue = asyncio.Queue()
    queue_iter = AsyncQueueIterator(queue, sentinel=None)

    # Starte Producer im Hintergrund
    producer_task = asyncio.create_task(producer(queue))
    
    print("Beginne Konsumierung ...")

    # Konsumiere mit anext()
    while True:
        try:
            item = await anext(queue_iter)
            print(f"Konsumiert: {item}")
        except StopAsyncIteration:
            print("Keine weiteren Elemente")
            break
    
    await producer_task


asyncio.run(consumer_with_anext())
Output 
Beginne Konsumierung ...
Konsumiert: Produziert: Item 0
Konsumiert: Produziert: Item 1
Konsumiert: Produziert: Item 2
Konsumiert: Produziert: Item 3
Konsumiert: Produziert: Item 4
Keine weiteren Elemente
/ Weiter

Zurück zu Builtin Functions

Zur Übersicht