Python

Asenkron

`AsyncOpenEmail`: `OpenEmail`'in her metodu, await ile, asyncio ya da trio üzerinde.

Asenkron istemci

AsyncOpenEmail, OpenEmail'in sahip olduğu her metoda aynı argümanlar ve aynı dönüş türleriyle sahiptir ve her biri await ile beklediğiniz bir coroutine'dir. Aynı anahtar sözcük argümanlarıyla oluşturulur, belirtmediğiniz her şey için aynı ortam değişkenlerini okur ve aynı hataları fırlatır.

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, blok ister normal biçimde ister bir istisnayla sona ersin, blok bittiğinde bağlantı havuzunu kapatır. Süreç boyunca yaşayan bir istemci başlangıçta bir kez oluşturulur ve kapanışta await client.aclose() ile kapatılır. Tüm program için tek bir istemci yeterlidir: olay döngüsündeki istediğiniz sayıda coroutine onu aynı anda kullanabilir.

Hazır openemail istemcisi ve init() senkrondur ve bunların asenkron ikizi yoktur. AsyncOpenEmail istemcinizi programın başladığı yerde oluşturun ve ihtiyaç duyan koda verin ya da framework'ünüzün uygulama durumunda tutun.

Paketin eşlik denetimi iki istemciyi metot metot karşılaştırır ve bir metot OpenEmail ile AsyncOpenEmail üzerinde farklı argümanlar aldığında başarısız olur; bu yüzden her metot sayfası ikisini birden anlatır.

async for ile sayfalama

list ve list_all, diğer tüm metotlar gibi await ile çağrılır. iterate öyle değildir: hemen bir asenkron yineleyici döndürür ve async for her sayfayı döngü ona ulaştığında getirir; bu yüzden döngüden çıkmak istekleri durdurur.

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

SayfalamaTek sayfa, tüm sayfalar ya da her seferinde tek öğe.

Aynı anda birçok çağrı

Tek bir istemci aynı anda istediğiniz sayıda isteği taşır. asyncio.gather onları birlikte başlatır ve bir semafor, aynı anda süren isteklerin sayısını seçtiğiniz bir sayıda tutar.

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 bugün sıradan okuma ve yazmaların hızına genel bir sınır koymuyor; bu yüzden ani bir yükü sizin için hiçbir şey yavaşlatmaz ve böyle bir sınır ileride eklenebilir. Saydığı şeyler ise 429 ile yanıtlanır: diğerlerinin yanı sıra çalışma alanının aylık gönderim kotası, günlük yapay zekâ işlemleri ve saatte 500 dosya yükleme. Bunların hiçbiri Retry-After taşımaz; bu yüzden istemci yeniden denemek yerine hemen is_rate_limited değeri true olan bir OpenEmailApiError fırlatır. return_exceptions=True, gather'ın ilk rette hata fırlatmak yerine her sonucu, reddedilen bir çağrıyı da kendi istisnası olarak döndürmesini sağlar ve döngü her birini okur.

Bir görevi iptal etmek isteğini de iptal eder ve yoldayken iptal edilen bir gönderim API'ye çoktan ulaşmış olabilir. İptal edip sonra tekrarlayabileceğiniz bir gönderime kendi idempotency_key= değerinizi verin; böylece tekrar, ikinci bir mesaj göndermek yerine ilkini yeniden oynatır.

Yeniden denemeler ve idempotencyİstemcinin neleri yeniden denediği ve yeniden denenen bir gönderimin neden tekrarlanamayacağı.HatalarNeyin hata fırlattığı ve neye göre dallanılacağı.

asyncio ve trio

İstemci bekleme ve zaman aşımı işlerini anyio, göndermeyi ise httpx üzerinden yapar ve ikisi de her iki olay döngüsünde çalışır; bu yüzden aynı main() hem asyncio.run(main()) hem de trio.run(main) altında çalışır. trio paketin bir bağımlılığı değildir; kullandığınızda onu kendiniz kurun.

Yukarıdaki örnekteki gibi asyncio'nun kendi gather ve Semaphore yapıları yalnızca asyncio üzerinde çalışır. Her ikisinde de çalışması gereken kod için, paketin zaten bağımlı olduğu anyio'nun create_task_group ve Semaphore yapılarını kullanın.

Asenkron bir erişim tokenı

Bir kişinin OAuth ile bağladığı bir uygulama, API anahtarı yerine bir erişim tokenı tutar ve onu access_token= olarak geçirir: ya tokenın kendisini ya da onu döndüren bir fonksiyonu. Fonksiyon her istekten önce çalışır; bu yüzden token süresinin dolmasına yaklaştığında onu yenileyebilir ve istemcinin hiç yeniden kurulması gerekmez. AsyncOpenEmail üzerinde bu bir async fonksiyon olabilir ve istemci onun döndürdüğünü await ile bekler.

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

Her istek onu beklediği için fonksiyonu hafif tutun: yukarıdaki gibi önbelleğe alınmış bir token döndürün ve onu yalnızca süresi dolmaya yakın yenileyin. OpenEmail bir coroutine'i bekleyemez; bu yüzden ona geçirilen bir async fonksiyon ilk istekte ValueError fırlatır.

YapılandırmaKimlik bilgileri, doğrulama kodları ve tüm seçenekler.

Tek kullanımlık gelen kutuları

create_async_temp_mail(), create_temp_mail()'in asenkron ikizidir. Hiç API anahtarı taşımaz: create ve list_domains hiçbir kimlik bilgisi göndermez, diğer her metot ise create çağrısının döndürdüğü tokenı inbox_token= olarak alır. Tek bir gelen kutusuna bağlı bir istemci için inbox_token= değerini doğrudan create_async_temp_mail fonksiyonuna geçirin.

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

Döndürdüğü istemcinin kendine ait bir aclose() metodu yoktur. İş bittiğinde bağlantılarını kapatmak için yukarıdaki gibi async with ile bir httpx.AsyncClient açın ve onu http_client= olarak geçirin.

Kendi httpx istemciniz

http_client=, bir proxy, bağlantı sınırları, kendi sertifikalarınız ya da testlerde sahte bir transport için oluşturduğunuz bir httpx.AsyncClient alır. Bir httpx.Client ise TypeError fırlatır, çünkü o OpenEmail içindir.

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

Geçirdiğiniz bir istemci sizin kalır: aclose() ve async with bloğunun sonu yalnızca SDK'nın açtığı bir havuzu kapatır; bu yüzden httpx.AsyncClient istemcinizi kendiniz kapatın, burada kendi async with bloğuyla. SDK'nın açtığı havuz yönlendirmeleri izler; aynı davranışı elde etmek için kendi istemcinizde follow_redirects=True ayarlayın. İstemcideki ya da tek bir çağrıdaki timeout=, httpx.AsyncClient hangi zaman aşımını taşırsa taşısın, her denemeyi yine sınırlar.

Testlerde httpx.MockTransport her isteği sizin bir fonksiyonunuzdan yanıtlar; böylece hiçbir şey ağa ulaşmaz.

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