8.12. Asynchrone Kontextmanager

Der Python-Überblick hat Kontextmanager vorgestellt – die Objekte, die with-Blöcke antreiben, wobei __enter__ auf dem Weg hinein und __exit__ auf dem Weg hinaus läuft, egal wie der Block endet. Asynchrone Kontextmanager sind dieselbe Idee, übertragen in asyncio: __aenter__ und __aexit__ sind Koroutinen, sodass das Einrichten oder das Aufräumen etwas awaiten kann. Der Block, der sie antreibt, ist async with.

Wir haben bereits einen verwendet. asyncio.Lock ist ein asynchroner Kontextmanager; die Seite locks hat ihn als async with bus_lock: ... verwendet. Auf dieser Seite geht es darum, einen für die eigene Ressource einer Anwendung zu schreiben.

8.12.1. Wann man einen schreibt

Wenn die Anwendung eine Ressource hat, die gepaartes Einrichten und Abbauen benötigt, und mindestens eine dieser Seiten etwas awaiten muss. Netzwerkverbindungen, Sensoren, die nach der Konfiguration eine Einschwingverzögerung brauchen, alles, was beim Eintritt etwas sperrt und beim Austritt entsperrt. Die schlichte synchrone with-Form ist nicht anwendbar, weil ihr __enter__ und __exit__ keine Koroutinen sein können.

8.12.2. Die zwei Methoden

  • async def __aenter__(self) läuft, wenn der Block betreten wird. Was auch immer sie zurückgibt, ist das, woran die optionale as name-Klausel von async with gebunden wird. self zurückzugeben ist die häufigste Form, aber jeder Wert funktioniert.

  • async def __aexit__(self, exc_type, exc, tb) läuft, wenn der Block verlassen wird. exc_type ist None bei einem normalen Austritt; bei einer Ausnahme (oder einem Abbruch) ist es die Klasse der Ausnahme, und exc ist die Instanz. Einen wahrheitsgemäßen Wert zurückzugeben teilt Python mit, dass die Ausnahme behandelt wurde und sich nicht fortpflanzen soll. None zurückzugeben (der übliche Fall) lässt die Ausnahme nach dem Aufräumen die Aufrufkette weiter hinauflaufen.

8.12.3. Ein durchgearbeitetes Beispiel

Ein Spotlight-Wrapper, der eine LED für den Rumpf des Blocks einschaltet, mit einer kurzen Einschwingverzögerung, damit die Beleuchtung stabil ist, bevor irgendwelche Aufnahmen stattfinden, und die LED auf dem Weg hinaus wieder ausschaltet:

import asyncio
from machine import LED


class Spotlight:
    def __init__(self, led):
        self._led = led

    async def __aenter__(self):
        self._led.on()
        await asyncio.sleep_ms(50)
        return self

    async def __aexit__(self, exc_type, exc, tb):
        self._led.off()

async def main():
    led = LED("LED_WHITE")

    async with Spotlight(led):
        # work that benefits from steady illumination
        await asyncio.sleep_ms(200)

asyncio.run(main())

Wenn der Block betreten wird, läuft __aenter__: Die LED geht an, das 50-ms-Einschwing-await gibt die Kontrolle an die Schleife ab, damit andere Koroutinen in der Zwischenzeit vorankommen können, und der Blockrumpf startet, sobald das Warten abgeschlossen ist. Wenn der Block endet – bei einem normalen Austritt, bei einer Ausnahme oder bei einem Abbruch – läuft __aexit__ und die LED geht wieder aus. Das Aufräumen läuft in jedem Fall; das ist die Garantie, die async with bietet.

The frame capture page shows how to make csi.CSI.snapshot() await-friendly; once that wrapper is in hand, the body of an async with Spotlight(led): block would typically be a capture loop running under the steady illumination.