비동기
`AsyncOpenEmail`: `OpenEmail`의 모든 메서드를 asyncio나 trio에서 await로 호출합니다.
비동기 클라이언트
AsyncOpenEmail은 OpenEmail이 가진 모든 메서드를 같은 인자와 같은 반환 타입으로 갖추고 있으며, 각 메서드는 await로 기다리는 코루틴입니다. 같은 키워드 인자로 만들고, 생략한 값은 같은 환경 변수에서 읽으며, 같은 오류를 발생시킵니다.
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는 루프가 각 페이지에 도달할 때 그 페이지를 가져오므로, 루프를 벗어나면 요청도 멈춥니다.
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는 요청을 한꺼번에 시작하고, 세마포어는 진행 중인 요청 수를 직접 정한 수로 제한합니다.
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로 응답합니다. 워크스페이스의 월간 발송 허용량, 하루 AI 작업, 시간당 500건의 파일 업로드 등이 그렇습니다. 이 중 어느 것도 Retry-After를 보내지 않으므로, 클라이언트는 재시도하지 않고 is_rate_limited가 true인 OpenEmailApiError를 즉시 발생시킵니다. return_exceptions=True를 주면 gather는 첫 거부에서 예외를 발생시키지 않고 모든 결과를 반환하며(거부된 호출은 그 예외가 결과가 됩니다), 루프가 각 결과를 읽습니다.
태스크를 취소하면 그 요청도 취소되지만, 전송 도중에 취소된 발송은 이미 API에 닿았을 수 있습니다. 취소했다가 다시 할 수도 있는 발송에는 직접 만든 idempotency_key=를 주십시오. 그러면 반복된 요청은 두 번째 메시지를 보내지 않고 첫 번째 발송을 재생합니다.
asyncio와 trio
클라이언트는 대기와 타임아웃을 anyio로, 전송을 httpx로 처리하며, 둘 다 어느 이벤트 루프에서든 동작하므로 같은 main()이 asyncio.run(main())에서도 trio.run(main)에서도 실행됩니다. trio는 패키지의 의존성이 아니므로, 사용할 때는 직접 설치하십시오.
위 예제처럼 asyncio 자체의 gather와 Semaphore는 asyncio에서만 동작합니다. 둘 다에서 실행되어야 하는 코드에는 패키지가 이미 의존하는 anyio의 create_task_group과 Semaphore를 쓰십시오.
비동기 액세스 토큰
사용자가 OAuth로 연결한 앱은 API 키 대신 액세스 토큰을 가지며, 이를 access_token=으로 넘깁니다. 토큰 자체이거나 토큰을 반환하는 함수입니다. 함수는 매 요청 전에 실행되므로, 만료가 가까워지면 토큰을 갱신할 수 있고 클라이언트를 다시 만들 필요가 없습니다. AsyncOpenEmail에서는 async 함수여도 되며, 클라이언트가 그 반환값을 await합니다.
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=으로 받습니다. 받은편지함 하나에 묶인 클라이언트를 만들려면 create_async_temp_mail 자체에 inbox_token=을 넘기십시오.
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()가 없습니다. 작업이 끝났을 때 연결을 닫으려면, 위 예제처럼 async with로 httpx.AsyncClient를 열고 그것을 http_client=로 넘기십시오.
직접 만든 httpx 클라이언트
http_client=는 프록시, 연결 수 제한, 직접 준비한 인증서, 테스트용 모의 전송 계층을 위해 직접 만든 httpx.AsyncClient를 받습니다. httpx.Client는 OpenEmail용이므로 TypeError를 발생시킵니다.
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가 직접 만든 함수로 모든 요청에 응답하므로, 아무것도 네트워크에 닿지 않습니다.
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())