42 lines
1.5 KiB
Python
42 lines
1.5 KiB
Python
"""Bounded Graph/mailbox HTTP concurrency for the upstream Email API."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
|
||
|
||
class GraphSemaphore:
|
||
"""Cap concurrent Graph-backed calls to one mailbox.
|
||
|
||
A free slot starts the next waiter immediately. On 429 the wait happens
|
||
outside the semaphore so in-flight stays under the cap; a retry only runs
|
||
after re-acquiring.
|
||
"""
|
||
|
||
def __init__(self,concurrency=4,max_retries=5):
|
||
self.concurrency=max(int(concurrency),1)
|
||
self.max_retries=max(int(max_retries),0)
|
||
self._slots=asyncio.Semaphore(self.concurrency)
|
||
|
||
def _retry_after_seconds(self,response,attempt):
|
||
raw=(response.headers.get("Retry-After") or "").strip()
|
||
if raw:
|
||
try:
|
||
return max(float(raw),0.1)
|
||
except ValueError:
|
||
pass
|
||
# ~1–2s base, doubles each attempt, soft cap so a wedged mailbox cannot sleep forever.
|
||
return min(1.5*(2**attempt),30.0)
|
||
|
||
async def get(self,client,url,*,params=None,headers=None):
|
||
"""GET under this semaphore; on 429 release, wait, then retry."""
|
||
attempt=0
|
||
while True:
|
||
async with self._slots:
|
||
response=await client.get(url,params=params,headers=headers)
|
||
if response.status_code!=429 or attempt>=self.max_retries:
|
||
return response
|
||
wait=self._retry_after_seconds(response,attempt)
|
||
await asyncio.sleep(wait)
|
||
attempt+=1
|