Конфигурация

В этих случаях ещё до отправки какого-либо запроса выбрасывается ValueError (или TypeError, где это указано в таблице), вместо того чтобы при вашей первой отправке всплыл непонятный сбой. В сообщении сказано, что было не так и что передать вместо этого.

Постройте клиент один раз и используйте его совместно. Новый экземпляр на каждый запрос без всякой пользы выбрасывает пул соединений и конфигурацию, а никакое состояние в нём не привязано к вызывающей стороне.

Один клиент можно безопасно использовать из нескольких потоков. close() или конец блока with закрывает пул соединений, который он открыл, а переданный вами http_client остаётся открытым, и закрыть его должны вы.

Для случая, который иначе потребовал бы по экземпляру на ключ (например, задания, отправляющего от имени нескольких рабочих пространств), передайте api_key в вызов. Он заменяет заголовок Authorization для этого запроса и ничего не оставляет на клиенте.

from openemail.types import EmailSend workspace_key = 'oe_live_...' message: EmailSend = {'from': sender, 'to': recipient, 'subject': subject, 'text': text} client.emails.send(message) client.emails.send(message, api_key=workspace_key) client.threads.list(folder='inbox', api_key=workspace_key)client.webhooks.list(api_key=workspace_key)

Каждый метод вне temp_mail принимает его как именованный аргумент рядом с timeout. Он проверяется до отправки запроса по тому же правилу, что и в конструкторе, так что опечатка выбросит ValueError с упоминанием api_key= on this call, а не 401 про учётные данные, которые потом придётся искать. Повторённый вызов сохраняет выданный ему ключ.

timeout задаётся в секундах и заменяет таймаут клиента для этого единственного вызова, на каждой его попытке.

client.raw.request отправляет запрос, применяя учётные данные, базовый URL, таймаут и политику повторов клиента, и возвращает разобранный JSON. Он принимает method, query, body, api_key и timeout. Он повторяет GET, а всё остальное отправляет один раз, если вы не передали repeatable=True. idempotent=True добавляет Idempotency-Key: тот, что вы передали как idempotency_key, или новый.

result = client.raw.request('/something-new', method='POST', body={'name': 'Invoices'})

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

from openemail import create_temp_mail temp_mail = create_temp_mail() inbox = temp_mail.create()messages = temp_mail.list_messages(inbox['id'], inbox_token=inbox['token']) print(messages['items'], messages['expiresAt'])

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

from openemail import OpenEmail client = OpenEmail(access_token=session.fresh_access_token) me = client.me.get() if me['object'] == 'oauth_token':    print(me['clientId'], me['expiresAt'])

Перед чувствительным изменением, например удалением домена или изменением вебхука, API спрашивает у токена доступа код подтверждения, который веб-приложение спросило бы у человека. Вызов выбрасывает OpenEmailApiError, у которого is_step_up_required равен True, и ничего не изменилось. Запросите код, подтвердите тот, что даст вам человек, затем сделайте вызов снова. У API-ключа код никогда не спрашивают.

from openemail import OpenEmailApiError try:    client.domains.delete(domain_id)except OpenEmailApiError as error:    if not error.is_step_up_required:        raise     challenge = client.security.begin_step_up()     if challenge['method'] == 'email':        prompt = f'Enter the code we emailed to {challenge.get("sentTo")}: '    else:        prompt = 'Enter the code from your authenticator app, or a backup code: '     client.security.verify_step_up({'code': input(prompt)})    client.domains.delete(domain_id)