غير متزامن
`AsyncOpenEmail`: كل دوال `OpenEmail`، تُستدعى مع await، على asyncio أو trio.
العميل غير المتزامن
يملك AsyncOpenEmail كل دالة يملكها OpenEmail، بالوسائط نفسها وأنواع الإرجاع نفسها، وكل منها coroutine تنتظرها بـ 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() عند الإيقاف. ويكفي عميل واحد للبرنامج كله: إذ يمكن لأي عدد من الـ coroutines على حلقة أحداثه استخدامه في الوقت نفسه.
العميل الجاهز 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 معًا، ويُبقي semaphore عدد الطلبات الجارية عند حدٍّ تختاره أنت.
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 في الحال بدلًا من إعادة المحاولة. ويجعل 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، وينتظر العميل ما تعيده.
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 عند أول طلب.
صناديق وارد مؤقتة
create_async_temp_mail() هو النظير غير المتزامن لـ create_temp_mail(). لا يحمل أي مفتاح API: إذ لا يرسل create وlist_domains أي اعتماد إطلاقًا، وتقبل كل دالة أخرى الرمز الذي أعاده create بوصفه inbox_token=. ومرّر inbox_token= إلى create_async_temp_mail نفسه للحصول على عميل مرتبط بصندوق واحد.
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 بنيته بنفسك، من أجل وكيل (proxy)، أو حدود للاتصالات، أو شهاداتك الخاصة، أو ناقل وهمي في الاختبارات. أما httpx.Client فيرفع TypeError، لأنه مخصص لـ OpenEmail.
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())