Asynchron
`AsyncOpenEmail`: jede Methode von `OpenEmail`, mit await aufgerufen, auf asyncio oder trio.
Der asynchrone Client
AsyncOpenEmail hat jede Methode, die OpenEmail hat, mit denselben Argumenten und denselben Rückgabetypen, und jede ist eine Koroutine, auf die Sie mit await warten. Er wird mit denselben Schlüsselwortargumenten erzeugt, liest für alles, was Sie weglassen, dieselben Umgebungsvariablen und löst dieselben Fehler aus.
import asyncio from openemail import AsyncOpenEmail async def main() -> None: async with AsyncOpenEmail() as client: sent = await client.emails.send({ 'from': 'Acme Billing <[email protected]>', 'to': '[email protected]', 'subject': 'Your September invoice', 'text': 'Your invoice is attached.', }) email = await client.emails.get(sent['id']) print(email['status'], email['sentAt']) asyncio.run(main())async with schließt den Verbindungspool, wenn der Block endet, ob regulär oder mit einer Ausnahme. Ein Client, der so lange lebt wie der Prozess, wird einmal beim Start erzeugt und beim Herunterfahren mit await client.aclose() geschlossen. Ein Client reicht für das ganze Programm: Beliebig viele Koroutinen auf seiner Event-Loop können ihn gleichzeitig nutzen.
Der fertige openemail-Client und init() sind synchron, und es gibt kein asynchrones Gegenstück dazu. Erzeugen Sie Ihren AsyncOpenEmail dort, wo das Programm startet, und reichen Sie ihn an den Code weiter, der ihn braucht, oder legen Sie ihn im Anwendungszustand Ihres Frameworks ab.
Die Paritätsprüfung des Pakets vergleicht die beiden Clients Methode für Methode und schlägt fehl, wenn eine Methode auf OpenEmail und AsyncOpenEmail unterschiedliche Argumente nimmt, jede Methodenseite beschreibt daher beide.
Paginieren mit async for
Auf list und list_all wird wie bei jeder anderen Methode mit await gewartet. Auf iterate nicht: Es gibt sofort einen asynchronen Iterator zurück, und async for holt jede Seite, wenn die Schleife sie erreicht, ein Verlassen der Schleife stoppt daher die Anfragen.
import asyncio from openemail import AsyncOpenEmail async def main() -> None: async with AsyncOpenEmail() as client: page = await client.emails.list(status='failed', limit=50) print(len(page['items']), page['nextCursor']) complaints = await client.suppressions.list_all(reason='complaint') print(len(complaints)) async for thread in client.threads.iterate(folder='inbox'): print(thread['id']) asyncio.run(main())Viele Aufrufe gleichzeitig
Ein Client trägt beliebig viele Anfragen gleichzeitig. asyncio.gather startet sie zusammen, und ein Semaphor begrenzt die Zahl der laufenden auf einen Wert, den Sie wählen.
import asyncio from openemail import AsyncOpenEmailfrom openemail.types import SentEmailResource async def main() -> None: recipients = ['[email protected]', '[email protected]', '[email protected]'] gate = asyncio.Semaphore(8) async with AsyncOpenEmail() as client: async def welcome(address: str) -> SentEmailResource: async with gate: return await client.emails.send({ 'from': 'Acme <[email protected]>', 'to': address, 'subject': 'Welcome to Acme', 'text': 'Your workspace is ready.', }) results = await asyncio.gather( *(welcome(address) for address in recipients), return_exceptions=True, ) for address, result in zip(recipients, results): if isinstance(result, BaseException): print(address, 'failed:', result) else: print(address, result['status']) asyncio.run(main())Die API setzt heute keine allgemeine Ratenbegrenzung für gewöhnliche Lese- und Schreibvorgänge, nichts bremst also eine Lastspitze für Sie aus, und eine solche Begrenzung kann später hinzukommen. Was sie zählt, beantwortet sie mit 429: unter anderem das monatliche Versandkontingent des Workspace, seine täglichen KI-Aktionen und 500 Datei-Uploads pro Stunde. Keine davon führt ein Retry-After mit, der Client löst daher sofort OpenEmailApiError mit is_rate_limited gleich true aus, statt es erneut zu versuchen. return_exceptions=True lässt gather jedes Ergebnis zurückgeben, einen abgelehnten Aufruf als seine Ausnahme, statt bei der ersten Ablehnung eine Ausnahme auszulösen, und die Schleife liest jedes einzeln.
Das Abbrechen eines Tasks bricht seine Anfrage ab, und ein Versand, der unterwegs abgebrochen wird, hat die API womöglich schon erreicht. Geben Sie einem Versand, den Sie abbrechen und dann wiederholen könnten, einen eigenen idempotency_key=, damit die Wiederholung den ersten erneut abspielt, statt eine zweite Nachricht zu senden.
asyncio und trio
Der Client wartet und überwacht Timeouts über anyio und sendet über httpx, und beide laufen auf jeder der beiden Event-Loops, sodass dasselbe main() sowohl unter asyncio.run(main()) als auch unter trio.run(main) läuft. trio ist keine Abhängigkeit des Pakets, installieren Sie es also selbst, wenn Sie es nutzen.
Die asyncio-eigenen gather und Semaphore, wie im Beispiel oben, laufen nur auf asyncio. Für Code, der auf beiden laufen muss, verwenden Sie create_task_group und Semaphore von anyio, von dem das Paket ohnehin abhängt.
Ein asynchrones Zugriffstoken
Eine App, die jemand über OAuth verbunden hat, hält ein Zugriffstoken statt eines API-Schlüssels und übergibt es als access_token=: das Token selbst oder eine Funktion, die es zurückgibt. Die Funktion läuft vor jeder Anfrage, sie kann das Token also erneuern, wenn es bald abläuft, und der Client muss nie neu gebaut werden. Auf AsyncOpenEmail darf sie eine async-Funktion sein, und der Client wartet mit await auf das, was sie zurückgibt.
import asyncioimport timefrom dataclasses import dataclass from openemail import AsyncOpenEmail from acme.auth import refresh_access_token @dataclassclass CachedToken: value: str = '' expires_at: float = 0.0 cached = CachedToken() async def access_token() -> str: if cached.expires_at - time.time() < 60: cached.value, lifetime = await refresh_access_token() cached.expires_at = time.time() + lifetime return cached.value async def main() -> None: async with AsyncOpenEmail(access_token=access_token) as client: me = await client.me.get() print(me['object']) asyncio.run(main())Halten Sie die Funktion günstig, denn jede Anfrage wartet auf sie: Geben Sie ein zwischengespeichertes Token zurück und erneuern Sie es erst kurz vor dem Ablauf, wie oben. OpenEmail kann nicht auf eine Koroutine warten, eine ihm übergebene async-Funktion löst daher bei der ersten Anfrage ValueError aus.
Wegwerf-Postfächer
create_async_temp_mail() ist das asynchrone Gegenstück zu create_temp_mail(). Der erzeugte Client trägt keinen API-Schlüssel: create und list_domains senden überhaupt keine Anmeldedaten, und jede andere Methode nimmt das Token, das create zurückgegeben hat, als inbox_token= entgegen. Übergeben Sie inbox_token= an create_async_temp_mail selbst, um einen an ein Postfach gebundenen Client zu erhalten.
import asyncio import httpxfrom openemail import create_async_temp_mail async def main() -> None: async with httpx.AsyncClient(follow_redirects=True) as http: temp = create_async_temp_mail(http_client=http) inbox = await temp.create({'ttlMinutes': 60}) print(inbox['address'], inbox['expiresAt']) async for message in temp.iterate_messages(inbox['id'], inbox_token=inbox['token']): print(message['from']['email'], message['subject']) asyncio.run(main())Der Client, den es zurückgibt, hat kein eigenes aclose(). Um seine Verbindungen nach getaner Arbeit zu schließen, öffnen Sie einen httpx.AsyncClient mit async with und übergeben ihn als http_client=, wie oben.
Ihr eigener httpx-Client
http_client= nimmt einen von Ihnen erzeugten httpx.AsyncClient entgegen, für einen Proxy, Verbindungslimits, eigene Zertifikate oder einen Mock-Transport in Tests. Ein httpx.Client löst TypeError aus, denn der ist für OpenEmail gedacht.
import asyncio import httpxfrom openemail import AsyncOpenEmail async def main() -> None: async with httpx.AsyncClient( proxy='http://proxy.internal:3128', limits=httpx.Limits(max_connections=20), follow_redirects=True, ) as http: client = AsyncOpenEmail(http_client=http, timeout=20) page = await client.threads.list(folder='inbox', limit=10) print(len(page['items'])) asyncio.run(main())Ein Client, den Sie übergeben, gehört weiterhin Ihnen: aclose() und das Ende von async with schließen nur einen Pool, den das SDK geöffnet hat, schließen Sie Ihren httpx.AsyncClient also selbst, hier mit seinem eigenen async with. Der Pool, den das SDK öffnet, folgt Weiterleitungen, setzen Sie daher auf Ihrem eigenen follow_redirects=True, damit sich beide gleich verhalten. timeout= auf dem Client oder bei einem einzelnen Aufruf begrenzt weiterhin jeden Versuch, ganz gleich, welches Timeout der httpx.AsyncClient mitführt.
In Tests beantwortet httpx.MockTransport jede Anfrage aus einer Funktion von Ihnen, sodass nichts das Netzwerk erreicht.
import asyncio import httpxfrom openemail import AsyncOpenEmail def answer(request: httpx.Request) -> httpx.Response: return httpx.Response(200, json={'object': 'list', 'data': [], 'hasMore': False, 'nextCursor': None}) async def main() -> None: async with httpx.AsyncClient(transport=httpx.MockTransport(answer)) as http: client = AsyncOpenEmail('oe_test_fixture', http_client=http) page = await client.suppressions.list() assert page['items'] == [] asyncio.run(main())