Python

Asíncrono

`AsyncOpenEmail`: todos los métodos de `OpenEmail`, con await, en asyncio o trio.

El cliente asíncrono

AsyncOpenEmail tiene todos los métodos que tiene OpenEmail, con los mismos argumentos y los mismos tipos de retorno, y cada uno es una corrutina que esperas con await. Se construye con los mismos argumentos nombrados, lee las mismas variables de entorno para todo lo que omitas y lanza los mismos errores.

async_client.py

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 cierra el pool de conexiones al terminar el bloque, tanto si termina con normalidad como con una excepción. Un cliente que vive tanto como el proceso se construye una vez al arrancar y se cierra con await client.aclose() al apagar. Un cliente basta para todo el programa: cualquier número de corrutinas de su bucle de eventos puede usarlo a la vez.

El cliente openemail ya preparado e init() son síncronos, y no tienen equivalente asíncrono. Construye tu AsyncOpenEmail donde arranca el programa y pásalo al código que lo necesite, o guárdalo en el estado de la aplicación de tu framework.

La comprobación de paridad del paquete compara los dos clientes método a método y falla cuando un método acepta argumentos distintos en OpenEmail y en AsyncOpenEmail, así que cada página de método describe ambos.

Paginar con async for

list y list_all se llaman con await como cualquier otro método. iterate no: devuelve un iterador asíncrono de inmediato, y async for obtiene cada página cuando el bucle llega a ella, así que salir del bucle detiene las solicitudes.

async_paging.py

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())

PaginaciónUna página, todas las páginas o un elemento cada vez.

Muchas llamadas a la vez

Un cliente admite cualquier número de solicitudes a la vez. asyncio.gather las inicia juntas, y un semáforo limita cuántas hay en curso al número que elijas.

gather.py

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())

Hoy la API no fija ningún límite general a la frecuencia de lecturas y escrituras ordinarias, así que nada frena automáticamente una ráfaga de solicitudes, y ese límite podría añadirse más adelante. Lo que sí contabiliza responde con un 429: la cuota mensual de envíos del espacio de trabajo, sus acciones de IA diarias y 500 subidas de archivos por hora, entre otros. Ninguno de ellos lleva un Retry-After, así que el cliente lanza OpenEmailApiError con is_rate_limited verdadero de inmediato en lugar de reintentar. return_exceptions=True hace que gather devuelva todos los resultados, con cada llamada rechazada como su excepción, en lugar de lanzar una excepción con el primer rechazo, y el bucle lee cada uno.

Cancelar una tarea cancela su solicitud, y un envío cancelado mientras estaba en curso puede haber llegado ya a la API. Dale tu propio idempotency_key= a un envío que quizá canceles y luego repitas, para que la repetición reproduzca el primero en lugar de enviar un segundo mensaje.

Reintentos e idempotenciaQué reintenta el cliente y por qué un envío reintentado no puede salir dos veces.ErroresQué lanza excepciones y qué comprobar para decidir.

asyncio y trio

El cliente hace sus pausas y aplica sus tiempos de espera con anyio y envía con httpx, y ambos funcionan en cualquiera de los dos bucles de eventos, así que el mismo main() se ejecuta con asyncio.run(main()) y con trio.run(main). trio no es una dependencia del paquete, así que instálalo tú mismo cuando lo uses.

Los propios gather y Semaphore de asyncio, como en el ejemplo anterior, solo funcionan en asyncio. Para código que tenga que funcionar en ambos, usa create_task_group y Semaphore de anyio, del que el paquete ya depende.

Un token de acceso asíncrono

Una app que una persona conectó con OAuth tiene un token de acceso en lugar de una clave de API, y lo pasa como access_token=: el propio token o una función que lo devuelva. La función se ejecuta antes de cada solicitud, así que puede renovar el token cuando esté a punto de caducar y nunca hay que reconstruir el cliente. En AsyncOpenEmail puede ser una función async, y el cliente espera con await lo que devuelve.

async_token.py

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())

Haz que la función sea barata, porque cada solicitud la espera: devuelve un token en caché y renuévalo solo cuando esté cerca de caducar, como arriba. OpenEmail no puede esperar a una corrutina, así que una función async que se le pase lanza ValueError en la primera solicitud.

ConfiguraciónCredenciales, códigos de verificación y todas las opciones.

Bandejas desechables

create_async_temp_mail() es el equivalente asíncrono de create_temp_mail(). No lleva ninguna clave de API: create y list_domains no envían ninguna credencial, y todos los demás métodos aceptan como inbox_token= el token que devolvió create. Pasa inbox_token= al propio create_async_temp_mail para obtener un cliente ligado a un buzón.

async_temp_mail.py

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())

El cliente que devuelve no tiene un aclose() propio. Para cerrar sus conexiones cuando termine el trabajo, abre un httpx.AsyncClient con async with y pásalo como http_client=, como arriba.

Tu propio cliente de httpx

http_client= acepta un httpx.AsyncClient que hayas construido tú, para un proxy, límites de conexiones, tus propios certificados o un transporte simulado en las pruebas. Un httpx.Client lanza TypeError, porque ese es para OpenEmail.

http_client.py

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 cliente que pasas sigue siendo tuyo: aclose() y el final de async with solo cierran un pool que haya abierto el SDK, así que cierra tú mismo tu httpx.AsyncClient, aquí con su propio async with. El pool que abre el SDK sigue las redirecciones, así que pon follow_redirects=True en el tuyo para que coincida. timeout= en el cliente, o en una sola llamada, sigue limitando cada intento, sea cual sea el tiempo de espera que lleve el httpx.AsyncClient.

En las pruebas, httpx.MockTransport responde a cada solicitud desde una función tuya, así que nada llega a la red.

test_with_mock.py

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())