---
title: "Async"
description: "`AsyncOpenEmail`: every method of `OpenEmail`, awaited, on asyncio or trio."
url: "https://openemail.uk/docs/python/async"
area: "Python"
category: "Getting started"
---

# 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.

**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 <billing@acme.com>',
            'to': 'ada@example.com',
            '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.

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

- [Pagination](https://openemail.uk/docs/python/pagination.md): One page, every page, or one item at a time.

## 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.

**gather.py**

```
import asyncio

from openemail import AsyncOpenEmail
from openemail.types import SentEmailResource

async def main() -> None:
    recipients = ['ada@example.com', 'grace@example.com', 'alan@example.com']
    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 <team@acme.com>',
                    '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.

- [Retries and idempotency](https://openemail.uk/docs/python/retries.md): What the client retries, and why a retried send cannot repeat.
- [Errors](https://openemail.uk/docs/python/errors.md): What raises, and what to branch on.

## 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.

**async_token.py**

```
import asyncio
import time
from dataclasses import dataclass

from openemail import AsyncOpenEmail

from acme.auth import refresh_access_token

@dataclass
class 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.

- [Configuration](https://openemail.uk/docs/python/configuration.md): Credentials, verification codes and every option.

## 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.

**async_temp_mail.py**

```
import asyncio

import httpx
from 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`.

**http_client.py**

```
import asyncio

import httpx
from 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.

**test_with_mock.py**

```
import asyncio

import httpx
from 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())
```