Python

ناهمگام

`AsyncOpenEmail`: همهٔ متدهای `OpenEmail`، با await، روی asyncio یا trio.

کلاینت ناهمگام

AsyncOpenEmail همهٔ متدهای OpenEmail را دارد، با همان آرگومان‌ها و همان نوع‌های بازگشتی، و هر کدام یک coroutine است که آن را await می‌کنید. با همان آرگومان‌های کلیدواژه‌ای ساخته می‌شود، برای هر چیزی که ندهید همان متغیرهای محیطی را می‌خواند، و همان خطاها را raise می‌کند.

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() بسته می‌شود. یک کلاینت برای کل برنامه کافی است: هر تعداد coroutine روی حلقهٔ رویداد آن می‌توانند هم‌زمان از آن استفاده کنند.

کلاینت آمادهٔ openemail و init() همگام‌اند، و همتای ناهمگامی ندارند. AsyncOpenEmail خودتان را جایی بسازید که برنامه آغاز می‌شود و به کدی بدهید که به آن نیاز دارد، یا آن را در وضعیت برنامه در فریم‌ورک خود نگه دارید.

بررسی همخوانیِ پکیج دو کلاینت را متد به متد مقایسه می‌کند و هر جا متدی روی OpenEmail و AsyncOpenEmail آرگومان‌های متفاوتی بگیرد شکست می‌خورد، پس صفحهٔ هر متد هر دو را توصیف می‌کند.

صفحه‌به‌صفحه با async for

list و list_all هم مثل هر متد دیگری با await فراخوانی می‌شوند. iterate نه: بی‌درنگ یک iterator ناهمگام برمی‌گرداند، و 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 آن‌ها را با هم آغاز می‌کند، و یک semaphore تعداد درخواست‌های در جریان را در عددی که خودتان انتخاب می‌کنید نگه می‌دارد.

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 امروز هیچ محدودیت عمومی‌ای بر نرخ خواندن‌ها و نوشتن‌های معمولی نمی‌گذارد، پس هیچ چیزی یک رگبار درخواست را برایتان کند نمی‌کند، و چنین محدودیتی ممکن است بعداً افزوده شود. آنچه واقعاً شمرده می‌شود با 429 پاسخ می‌دهد: از جمله سهمیهٔ ارسال ماهانهٔ فضای کاری، کنش‌های هوش مصنوعی روزانه‌اش و 500 بارگذاری فایل در ساعت. هیچ‌کدام از این‌ها Retry-After ندارند، پس کلاینت به‌جای تلاش دوباره، بی‌درنگ OpenEmailApiError را با is_rate_limited برابر true raise می‌کند. return_exceptions=True باعث می‌شود gather به‌جای raise کردن در نخستین رد شدن، همهٔ نتیجه‌ها را برگرداند (فراخوانی ردشده را به شکل استثنایش)، و حلقه هر کدام را می‌خواند.

لغو یک task درخواستش را لغو می‌کند، و ارسالی که در میانهٔ راه لغو شود ممکن است پیش‌تر به API رسیده باشد. به ارسالی که ممکن است لغوش کنید و سپس تکرارش کنید یک idempotency_key= از خودتان بدهید، تا تکرار، همان ارسال نخست را بازپخش کند و پیام دومی نفرستد.

تلاش مجدد و idempotencyکلاینت چه چیزی را دوباره تلاش می‌کند، و چرا ارسالی که دوباره تلاش شود تکرار نمی‌شود.خطاهاچه چیزی خطا raise می‌کند، و روی چه چیزی شاخه بزنید.

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 نمی‌تواند منتظر یک coroutine بماند، پس تابع async ای که به آن داده شود در نخستین درخواست ValueError را raise می‌کند.

پیکربندیاعتبارنامه‌ها، کدهای تأیید هویت و همهٔ گزینه‌ها.

صندوق‌های یک‌بارمصرف

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 را می‌گیرد که خودتان ساخته‌اید، برای یک پراکسی، محدودیت اتصال‌ها، گواهی‌های خودتان یا یک transport ساختگی در آزمون‌ها. یک httpx.Client خطای TypeError را raise می‌کند، چون آن یکی برای 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 باز می‌کند redirectها را دنبال می‌کند، پس برای هم‌خوانی، 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())