Asynchrone
`AsyncOpenEmail` : toutes les méthodes d'`OpenEmail`, avec await, sur asyncio ou trio.
Le client asynchrone
AsyncOpenEmail a toutes les méthodes d'OpenEmail, avec les mêmes arguments et les mêmes types de retour, et chacune est une coroutine que vous attendez avec await. Il se construit avec les mêmes arguments nommés, lit les mêmes variables d'environnement pour tout ce que vous omettez, et lève les mêmes erreurs.
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 ferme le pool de connexions à la fin du bloc, que celui-ci se termine normalement ou par une exception. Un client qui vit aussi longtemps que le processus se construit une fois au démarrage et se ferme avec await client.aclose() à l'arrêt. Un seul client suffit pour tout le programme : n'importe quel nombre de coroutines sur sa boucle d'événements peut l'utiliser en même temps.
Le client openemail prêt à l'emploi et init() sont synchrones, et ils n'ont pas d'équivalent asynchrone. Construisez votre AsyncOpenEmail là où le programme démarre et transmettez-le au code qui en a besoin, ou gardez-le dans l'état d'application de votre framework.
Le contrôle de parité du paquet compare les deux clients méthode par méthode et échoue quand une méthode prend des arguments différents sur OpenEmail et sur AsyncOpenEmail : chaque page de méthode décrit donc les deux.
Paginer avec async for
list et list_all s'appellent avec await, comme toutes les autres méthodes. Pas iterate : il renvoie immédiatement un itérateur asynchrone, et async for récupère chaque page quand la boucle l'atteint, si bien que sortir de la boucle arrête les requêtes.
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())Plusieurs appels à la fois
Un seul client mène n'importe quel nombre de requêtes à la fois. asyncio.gather les lance ensemble, et un sémaphore limite le nombre de requêtes en cours à la valeur que vous choisissez.
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())L'API n'impose aujourd'hui aucune limite générale de débit sur les lectures et écritures ordinaires : rien ne ralentit donc une rafale à votre place, et une telle limite pourrait être ajoutée plus tard. Ce qu'elle compte bel et bien répond 429 : le quota mensuel d'envois de l'espace de travail, ses actions d'IA quotidiennes et 500 téléversements de fichiers par heure, entre autres. Aucune de ces réponses ne porte de Retry-After : le client lève donc immédiatement OpenEmailApiError avec is_rate_limited à vrai au lieu de réessayer. return_exceptions=True fait renvoyer à gather chaque résultat, un appel refusé sous forme de son exception, au lieu de lever une exception au premier refus, et la boucle lit chacun d'eux.
Annuler une tâche annule sa requête, et un envoi annulé en cours de route peut avoir déjà atteint l'API. Donnez à un envoi que vous risquez d'annuler puis de répéter votre propre idempotency_key=, pour que la répétition rejoue le premier au lieu d'envoyer un second message.
asyncio et trio
Le client gère ses attentes et ses timeouts via anyio et envoie via httpx, et les deux fonctionnent sur l'une comme sur l'autre boucle d'événements : le même main() tourne donc sous asyncio.run(main()) et sous trio.run(main). trio n'est pas une dépendance du paquet : installez-le vous-même si vous l'utilisez.
gather et Semaphore propres à asyncio, comme dans l'exemple ci-dessus, ne fonctionnent que sur asyncio. Pour du code qui doit tourner sur les deux, utilisez create_task_group et Semaphore d'anyio, dont le paquet dépend déjà.
Un jeton d'accès asynchrone
Une application qu'une personne a connectée avec OAuth détient un jeton d'accès plutôt qu'une clé API, et le passe comme access_token= : le jeton lui-même, ou une fonction qui le renvoie. La fonction s'exécute avant chaque requête : elle peut donc renouveler le jeton quand il approche de son expiration, et le client n'a jamais à être reconstruit. Sur AsyncOpenEmail, ce peut être une fonction async, et le client attend ce qu'elle renvoie.
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())Gardez la fonction peu coûteuse, puisque chaque requête l'attend : renvoyez un jeton en cache et ne le renouvelez qu'à l'approche de l'expiration, comme ci-dessus. OpenEmail ne peut pas attendre une coroutine : une fonction async qui lui est passée lève donc ValueError à la première requête.
Boîtes jetables
create_async_temp_mail() est l'équivalent asynchrone de create_temp_mail(). Il ne porte aucune clé API : create et list_domains n'envoient aucun identifiant, et toutes les autres méthodes prennent le jeton renvoyé par create comme inbox_token=. Passez inbox_token= à create_async_temp_mail lui-même pour obtenir un client lié à une seule boîte.
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())Le client qu'il renvoie n'a pas d'aclose() à lui. Pour fermer ses connexions une fois le travail terminé, ouvrez un httpx.AsyncClient avec async with et passez-le comme http_client=, comme ci-dessus.
Votre propre client httpx
http_client= accepte un httpx.AsyncClient que vous avez construit, pour un proxy, des limites de connexions, vos propres certificats ou un transport factice dans les tests. Un httpx.Client lève TypeError, puisque celui-là est destiné à OpenEmail.
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())Un client que vous passez reste le vôtre : aclose() et la fin d'async with ne ferment qu'un pool ouvert par le SDK. Fermez donc vous-même votre httpx.AsyncClient, ici avec son propre async with. Le pool qu'ouvre le SDK suit les redirections : réglez follow_redirects=True sur le vôtre pour obtenir le même comportement. timeout= sur le client, ou sur un appel isolé, borne toujours chaque tentative, quel que soit le timeout que porte le httpx.AsyncClient.
Dans les tests, httpx.MockTransport répond à chaque requête à partir d'une de vos fonctions : rien n'atteint donc le réseau.
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())