Python

Assíncrono

`AsyncOpenEmail`: todos os métodos de `OpenEmail`, com await, em asyncio ou trio.

O cliente assíncrono

AsyncOpenEmail tem todos os métodos que OpenEmail tem, com os mesmos argumentos e os mesmos tipos de retorno, e cada um é uma corrotina que aguarda com await. É construído com os mesmos argumentos nomeados, lê as mesmas variáveis de ambiente para tudo o que omitir e lança os mesmos erros.

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 fecha o pool de ligações quando o bloco termina, quer termine normalmente quer com uma exceção. Um cliente que vive tanto quanto o processo é construído uma vez no arranque e fechado com await client.aclose() no encerramento. Um cliente chega para o programa inteiro: qualquer número de corrotinas no seu ciclo de eventos pode usá-lo ao mesmo tempo.

O cliente openemail já pronto e init() são síncronos, e não têm equivalente assíncrono. Construa o seu AsyncOpenEmail onde o programa arranca e passe-o ao código que precisa dele, ou guarde-o no estado da aplicação do seu framework.

A verificação de paridade do pacote compara os dois clientes método a método e falha quando um método aceita argumentos diferentes em OpenEmail e em AsyncOpenEmail, por isso cada página de método descreve ambos.

Paginar com async for

list e list_all são chamados com await como qualquer outro método. iterate não: devolve logo um iterador assíncrono, e async for obtém cada página quando o ciclo lá chega, por isso sair do ciclo interrompe os pedidos.

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

PaginaçãoUma página, todas as páginas ou um item de cada vez.

Muitas chamadas ao mesmo tempo

Um cliente aguenta qualquer número de pedidos ao mesmo tempo. asyncio.gather inicia-os em conjunto, e um semáforo limita quantos estão em curso ao número que escolher.

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

Hoje a API não define nenhum limite geral à taxa de leituras e escritas comuns, por isso nada abranda automaticamente uma rajada de pedidos, e esse limite pode vir a ser acrescentado mais tarde. O que ela contabiliza responde com 429: a quota mensal de envios do espaço de trabalho, as suas ações de IA diárias e 500 carregamentos de ficheiros por hora, entre outros. Nenhum deles traz um Retry-After, por isso o cliente lança OpenEmailApiError com is_rate_limited verdadeiro de imediato, em vez de repetir. return_exceptions=True faz com que gather devolva todos os resultados, com cada chamada recusada como a sua exceção, em vez de lançar uma exceção à primeira recusa, e o ciclo lê cada um.

Cancelar uma tarefa cancela o seu pedido, e um envio cancelado a meio pode já ter chegado à API. Dê o seu próprio idempotency_key= a um envio que possa cancelar e depois repetir, para que a repetição reproduza o primeiro em vez de enviar uma segunda mensagem.

Repetições e idempotênciaO que o cliente volta a tentar, e porque é que um envio repetido não pode sair duas vezes.ErrosO que lança exceções, e o que verificar para decidir.

asyncio e trio

O cliente faz as suas pausas e aplica os seus timeouts através do anyio e envia através do httpx, e ambos correm em qualquer um dos ciclos de eventos, por isso o mesmo main() corre com asyncio.run(main()) e com trio.run(main). O trio não é uma dependência do pacote, por isso instale-o você quando o usar.

Os próprios gather e Semaphore do asyncio, como no exemplo acima, só correm em asyncio. Para código que tenha de correr em ambos, use create_task_group e Semaphore do anyio, de que o pacote já depende.

Um token de acesso assíncrono

Uma aplicação que alguém ligou com OAuth tem um token de acesso em vez de uma chave de API, e passa-o como access_token=: o próprio token ou uma função que o devolve. A função é executada antes de cada pedido, por isso pode renovar o token quando estiver perto de expirar e o cliente nunca tem de ser reconstruído. Em AsyncOpenEmail pode ser uma função async, e o cliente aguarda com await o que ela devolve.

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

Mantenha a função leve, porque cada pedido espera por ela: devolva um token em cache e renove-o só perto de expirar, como acima. OpenEmail não consegue esperar por uma corrotina, por isso uma função async que lhe seja passada lança ValueError no primeiro pedido.

ConfiguraçãoCredenciais, códigos de verificação e todas as opções.

Caixas de entrada descartáveis

create_async_temp_mail() é o equivalente assíncrono de create_temp_mail(). Não transporta nenhuma chave de API: create e list_domains não enviam credencial nenhuma, e todos os outros métodos recebem como inbox_token= o token que create devolveu. Passe inbox_token= ao próprio create_async_temp_mail para ter um cliente associado a uma caixa de entrada.

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

O cliente que devolve não tem um aclose() próprio. Para fechar as suas ligações quando o trabalho terminar, abra um httpx.AsyncClient com async with e passe-o como http_client=, como acima.

O seu próprio cliente httpx

http_client= aceita um httpx.AsyncClient construído por si, para um proxy, limites de ligações, os seus próprios certificados ou um transporte simulado nos testes. Um httpx.Client lança TypeError, porque esse é 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())

Um cliente que passa continua a ser seu: aclose() e o fim de async with só fecham um pool que o SDK tenha aberto, por isso feche você o seu httpx.AsyncClient, aqui com o seu próprio async with. O pool que o SDK abre segue redirecionamentos, por isso defina follow_redirects=True no seu para corresponder. timeout= no cliente, ou numa única chamada, continua a limitar cada tentativa, seja qual for o timeout que o httpx.AsyncClient traga.

Nos testes, httpx.MockTransport responde a cada pedido a partir de uma função sua, por isso nada chega à rede.

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