Python

Асинхронность

`AsyncOpenEmail`: все методы `OpenEmail` через await, на asyncio или trio.

Асинхронный клиент

У AsyncOpenEmail есть все методы OpenEmail с теми же аргументами и теми же типами возвращаемых значений, и каждый из них является корутиной, которую вы ждёте через await. Он создаётся с теми же именованными аргументами, читает те же переменные окружения для всего, что вы не указали, и выбрасывает те же ошибки.

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 закрывает пул соединений, когда блок заканчивается, нормально или с исключением. Клиент, который живёт столько же, сколько процесс, строится один раз при запуске и закрывается через await client.aclose() при остановке. Одного клиента достаточно на всю программу: им может одновременно пользоваться любое число корутин в его цикле событий.

Готовый клиент openemail и init() синхронные, и асинхронного аналога у них нет. Постройте свой AsyncOpenEmail там, где стартует программа, и передайте его коду, которому он нужен, или храните его в состоянии приложения вашего фреймворка.

Проверка соответствия в пакете сравнивает два клиента метод за методом и падает, когда метод принимает разные аргументы в OpenEmail и AsyncOpenEmail, поэтому каждая страница метода описывает оба.

Постраничный обход с async for

list и list_all вызываются через await, как и все остальные методы. С iterate не так: он сразу возвращает асинхронный итератор, а async for запрашивает каждую страницу, когда цикл до неё доходит, так что выход из цикла прекращает запросы.

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

ПагинацияОдна страница, все страницы или по одному элементу.

Много вызовов одновременно

Один клиент ведёт любое число запросов одновременно. asyncio.gather запускает их вместе, а семафор ограничивает число выполняющихся запросов выбранным вами значением.

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

Сегодня API не задаёт общего ограничения частоты обычных чтений и записей, так что ничто не замедлит всплеск запросов за вас, и такое ограничение может появиться позже. То, что API всё же считает, отвечает 429: месячный лимит отправок рабочего пространства, его дневные действия ИИ и 500 загрузок файлов в час, среди прочего. Ни один из этих ответов не несёт Retry-After, поэтому клиент сразу выбрасывает OpenEmailApiError с истинным is_rate_limited, а не повторяет запрос. return_exceptions=True заставляет gather вернуть все результаты, отклонённый вызов в виде его исключения, вместо того чтобы выбросить исключение при первом отказе, а цикл читает каждый из них.

Отмена задачи отменяет её запрос, а отправка, отменённая на лету, могла уже дойти до API. Дайте отправке, которую вы можете отменить и затем повторить, собственный idempotency_key=, чтобы повтор воспроизвёл первую отправку, а не отправил второе письмо.

Повторы и идемпотентностьЧто клиент повторяет и почему повторённая отправка не может задвоиться.ОшибкиЧто выбрасывает исключения и по чему ветвиться.

asyncio и trio

Клиент ждёт и отсчитывает таймауты через anyio, а отправляет через httpx, и оба работают на любом из двух циклов событий, так что один и тот же main() запускается и под asyncio.run(main()), и под trio.run(main). trio не является зависимостью пакета, поэтому установите его сами, если используете.

Собственные gather и Semaphore из asyncio, как в примере выше, работают только на asyncio. Для кода, который должен работать на обоих, используйте create_task_group и Semaphore из anyio, от которого пакет уже зависит.

Асинхронный токен доступа

Приложение, которое человек подключил через OAuth, держит токен доступа вместо API-ключа и передаёт его как access_token=: сам токен или функцию, которая его возвращает. Функция выполняется перед каждым запросом, поэтому может обновить токен, когда срок его действия подходит к концу, и клиент никогда не придётся пересоздавать. В AsyncOpenEmail это может быть функция async, и клиент дожидается того, что она вернёт.

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

Делайте функцию дешёвой, ведь её ждёт каждый запрос: возвращайте закэшированный токен и обновляйте его только ближе к истечению, как выше. OpenEmail не может ждать корутину, поэтому переданная ему функция async выбрасывает ValueError на первом запросе.

КонфигурацияУчётные данные, коды подтверждения и все опции.

Одноразовые ящики

create_async_temp_mail() является асинхронным аналогом create_temp_mail(). Он не несёт API-ключа: create и list_domains вообще не отправляют учётных данных, а каждый другой метод принимает токен, который вернул create, как inbox_token=. Передайте inbox_token= самому create_async_temp_mail, чтобы получить клиент, привязанный к одному ящику.

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

У клиента, который он возвращает, нет собственного aclose(). Чтобы закрыть его соединения по окончании работы, откройте httpx.AsyncClient через async with и передайте его как http_client=, как выше.

Собственный клиент httpx

http_client= принимает построенный вами httpx.AsyncClient: для прокси, лимитов соединений, собственных сертификатов или фиктивного транспорта в тестах. httpx.Client выбрасывает TypeError, потому что он предназначен для 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())

Переданный вами клиент остаётся вашим: aclose() и конец async with закрывают только пул, который открыл SDK, так что закрывайте свой httpx.AsyncClient сами, здесь его собственным async with. Пул, который открывает SDK, следует перенаправлениям, поэтому задайте follow_redirects=True у своего, чтобы поведение совпадало. timeout= на клиенте или на отдельном вызове по-прежнему ограничивает каждую попытку, какой бы таймаут ни был у httpx.AsyncClient.

В тестах httpx.MockTransport отвечает на каждый запрос из вашей функции, так что в сеть ничего не уходит.

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