Python

Asinkron

`AsyncOpenEmail`: çdo metodë e `OpenEmail`, e thirrur me await, në asyncio ose trio.

Klienti asinkron

AsyncOpenEmail ka çdo metodë që ka OpenEmail, me të njëjtat argumente dhe të njëjtat tipe kthimi, dhe secila është një korutinë që e prisni me await. Ndërtohet me të njëjtat argumente me fjalë kyçe, lexon të njëjtat variabla mjedisi për çdo gjë që lini jashtë dhe ngre të njëjtat gabime.

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 e mbyll pool-in e lidhjeve kur mbaron blloku, qoftë normalisht, qoftë me një përjashtim. Një klient që jeton sa procesi ndërtohet një herë në nisje dhe mbyllet me await client.aclose() në fikje. Një klient mjafton për gjithë programin: çdo numër korutinash në ciklin e tij të ngjarjeve mund ta përdorë njëkohësisht.

Klienti i gatshëm openemail dhe init() janë sinkronë, dhe nuk kanë binjak asinkron. Ndërtojeni AsyncOpenEmail-in tuaj aty ku niset programi dhe jepjani kodit që ka nevojë për të, ose mbajeni në gjendjen e aplikacionit të framework-ut tuaj.

Kontrolli i përputhshmërisë i paketës i krahason të dy klientët metodë për metodë dhe dështon kur një metodë merr argumente të ndryshme te OpenEmail dhe AsyncOpenEmail, ndaj çdo faqe metode i përshkruan të dy.

Faqosja me async for

list dhe list_all thirren me await si çdo metodë tjetër. iterate jo: ajo kthen menjëherë një iterator asinkron, dhe async for merr çdo faqe kur cikli arrin tek ajo, kështu që dalja nga cikli i ndal kërkesat.

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

FaqosjaNjë faqe, të gjitha faqet, ose një element në një kohë.

Shumë thirrje njëherësh

Një klient mban çdo numër kërkesash njëherësh. asyncio.gather i nis bashkë, dhe një semafor e kufizon numrin e atyre që janë në ecuri te një shifër që zgjidhni ju.

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

Sot API-ja nuk vendos asnjë kufi të përgjithshëm shpejtësie për leximet dhe shkrimet e zakonshme, ndaj asgjë nuk e ngadalëson për ju një shpërthim kërkesash, dhe një kufi i tillë mund të shtohet më vonë. Ajo që numëron përgjigjet me 429: ndër të tjera, kuota mujore e dërgimeve të hapësirës së punës, veprimet e saj ditore AI dhe 500 ngarkime skedarësh në orë. Asnjëra prej tyre nuk mbart Retry-After, ndaj klienti ngre menjëherë OpenEmailApiError me is_rate_limited true, në vend që të riprovojë. return_exceptions=True bën që gather të kthejë çdo rezultat, një thirrje të refuzuar si përjashtimin e saj, në vend që të ngrejë gabim te refuzimi i parë, dhe cikli lexon secilin.

Anulimi i një detyre anulon kërkesën e saj, dhe një dërgim i anuluar gjatë rrugës mund të ketë arritur tashmë te API-ja. Një dërgimi që mund ta anuloni e pastaj ta përsërisni jepini një idempotency_key= tuajin, kështu që përsëritja e riluan të parin në vend që të dërgojë një mesazh të dytë.

Riprovat dhe idempotencaÇfarë riprovon klienti dhe pse një dërgim i riprovuar nuk mund të përsëritet.GabimetÇfarë ngre përjashtime dhe sipas çfarë të degëzoni.

asyncio dhe trio

Klienti pret dhe mat afatet kohore përmes anyio dhe dërgon përmes httpx, dhe të dyja funksionojnë në cilindo cikël ngjarjesh, kështu që i njëjti main() ekzekutohet si nën asyncio.run(main()), ashtu edhe nën trio.run(main). trio nuk është varësi e paketës, ndaj instalojeni vetë kur e përdorni.

gather dhe Semaphore të vetë asyncio-s, si në shembullin më sipër, funksionojnë vetëm në asyncio. Për kod që duhet të funksionojë në të dyja, përdorni create_task_group dhe Semaphore të anyio-s, nga e cila paketa varet tashmë.

Një token qasjeje asinkron

Një aplikacion që dikush e lidhi me OAuth mban një token qasjeje në vend të një çelësi API dhe e jep si access_token=: vetë tokenin, ose një funksion që e kthen atë. Funksioni ekzekutohet para çdo kërkese, ndaj mund ta rinovojë tokenin kur i afrohet skadimit dhe klienti nuk ka nevojë të rindërtohet kurrë. Te AsyncOpenEmail ai mund të jetë një funksion async, dhe klienti e pret me await atë që ai kthen.

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

Mbajeni funksionin të lehtë, sepse çdo kërkesë e pret: ktheni një token të ruajtur në cache dhe rinovojeni vetëm afër skadimit, si më sipër. OpenEmail nuk mund të presë një korutinë, ndaj një funksion async që i jepet atij ngre ValueError te kërkesa e parë.

KonfigurimiKredencialet, kodet e verifikimit dhe çdo opsion.

Kuti të përkohshme

create_async_temp_mail() është binjaku asinkron i create_temp_mail(). Nuk mban asnjë çelës API: create dhe list_domains nuk dërgojnë asnjë kredencial fare, dhe çdo metodë tjetër merr si inbox_token= token-in që ktheu create. Jepjani inbox_token= vetë create_async_temp_mail për një klient të lidhur me një kuti të vetme.

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

Klienti që kthen ajo nuk ka aclose() të vetin. Për t'i mbyllur lidhjet e tij kur të mbarojë puna, hapni një httpx.AsyncClient me async with dhe jepeni si http_client=, si më sipër.

Klienti juaj httpx

http_client= merr një httpx.AsyncClient që e ndërtuat ju, për një proxy, kufij lidhjesh, certifikatat tuaja ose një transport mock në teste. Një httpx.Client ngre TypeError, sepse ai është për 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())

Një klient që jepni mbetet i juaji: aclose() dhe fundi i async with mbyllin vetëm një pool që hapi SDK-ja, ndaj mbylleni vetë httpx.AsyncClient-in tuaj, këtu me async with-in e vet. Pool-i që hap SDK-ja ndjek ridrejtimet, ndaj vendosni follow_redirects=True te i juaji që të sillen njësoj. timeout= te klienti, ose te një thirrje e vetme, vazhdon ta kufizojë çdo përpjekje, cilido qoftë timeout-i që mbart httpx.AsyncClient.

Në teste, httpx.MockTransport i përgjigjet çdo kërkese nga një funksion i juaji, kështu që asgjë nuk arrin në rrjet.

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