एसिंक

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

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

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

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

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

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

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