Async
`AsyncOpenEmail`: every method of `OpenEmail`, awaited, on asyncio or trio.
The async client
AsyncOpenEmail has every method OpenEmail has, with the same arguments and the same return types, and each one is a coroutine you await. It is built with the same keyword arguments, reads the same environment variables for anything you leave out, and raises the same errors.
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 closes the connection pool when the block ends, whether it ends normally or with an exception. A client that lives as long as the process is built once at startup and closed with await client.aclose() at shutdown. One client is enough for the whole program: any number of coroutines on its event loop can use it at the same time.
The ready made openemail client and init() are synchronous, and there is no async twin of them. Build your AsyncOpenEmail where the program starts and hand it to the code that needs it, or keep it on your framework’s application state.
The package’s parity check compares the two clients method by method and fails when a method takes different arguments on OpenEmail and AsyncOpenEmail, so every method page describes both.
Paging with async for
list and list_all are awaited like every other method. iterate is not: it returns an async iterator straight away, and async for fetches each page when the loop reaches it, so breaking out of the loop stops the requests.
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())Many calls at once
One client carries any number of requests at once. asyncio.gather starts them together, and a semaphore holds how many are in flight to a number you choose.
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())The API sets no general limit on the rate of ordinary reads and writes today, so nothing slows a burst down for you, and such a limit may be added later. What it does count answers 429: the monthly send allowance of the workspace, its daily AI actions and 500 file uploads an hour, among others. None of these carries a Retry-After, so the client raises OpenEmailApiError with is_rate_limited true at once rather than retrying. return_exceptions=True makes gather return every outcome, a refused call as its exception, instead of raising at the first refusal, and the loop reads each one.
Cancelling a task cancels its request, and a send cancelled in flight may already have reached the API. Give a send you might cancel and then repeat an idempotency_key= of your own, so the repeat replays the first one rather than sending a second message.
asyncio and trio
The client sleeps and times out through anyio and sends through httpx, and both run on either event loop, so the same main() runs under asyncio.run(main()) and under trio.run(main). trio is not a dependency of the package, so install it yourself when you use it.
asyncio’s own gather and Semaphore, as in the example above, run only on asyncio. For code that has to run on both, use anyio’s create_task_group and Semaphore, which the package already depends on.
An async access token
An app a person connected with OAuth holds an access token rather than an API key, and passes it as access_token=: the token itself, or a function that returns it. The function runs before every request, so it can renew the token when it is close to expiring and the client never has to be rebuilt. On AsyncOpenEmail it may be an async function, and the client awaits what it returns.
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())Keep the function cheap, since every request waits for it: return a cached token and renew it only near expiry, as above. OpenEmail cannot wait for a coroutine, so an async function passed to it raises ValueError on the first request.
Disposable inboxes
create_async_temp_mail() is the async twin of create_temp_mail(). It carries no API key: create and list_domains send no credential at all, and every other method takes the token create returned as inbox_token=. Pass inbox_token= to create_async_temp_mail itself for a client bound to one inbox.
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())The client it returns has no aclose() of its own. To close its connections when the work is done, open an httpx.AsyncClient with async with and pass it as http_client=, as above.
Your own httpx client
http_client= takes an httpx.AsyncClient you built, for a proxy, connection limits, your own certificates or a mock transport in tests. An httpx.Client raises TypeError, since that one is for 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())A client you pass stays yours: aclose() and the end of async with close only a pool the SDK opened, so close your httpx.AsyncClient yourself, here with its own async with. The pool the SDK opens follows redirects, so set follow_redirects=True on yours to match. timeout= on the client, or on a single call, still bounds every attempt, whatever timeout the httpx.AsyncClient carries.
In tests, httpx.MockTransport answers every request from a function of yours, so nothing reaches the network.
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())