diff --git a/app/api/v1/endpoints/filings.py b/app/api/v1/endpoints/filings.py index 7fc37d0..27cb51c 100644 --- a/app/api/v1/endpoints/filings.py +++ b/app/api/v1/endpoints/filings.py @@ -314,7 +314,16 @@ async def get_exhibit_bulk( error=str(e), ) - raw = await asyncio.gather(*[_fetch_one(item) for item in request.items], return_exceptions=True) + sem = asyncio.Semaphore(4) + + async def _fetch_one_limited(item): + async with sem: + return await _fetch_one(item) + + raw = await asyncio.wait_for( + asyncio.gather(*[_fetch_one_limited(item) for item in request.items], return_exceptions=True), + timeout=300, + ) results: List[BulkExhibitItem] = [] for r in raw: if isinstance(r, Exception): diff --git a/app/services/sec_http_client.py b/app/services/sec_http_client.py index b52f2ef..02fd391 100644 --- a/app/services/sec_http_client.py +++ b/app/services/sec_http_client.py @@ -6,6 +6,7 @@ Extracted from etf_holdings_fetcher.py to be reused by all SEC-related services. import time as _time import asyncio +import contextvars import aiohttp import hashlib import json @@ -16,6 +17,11 @@ from typing import Dict, Optional from app.core.config import settings +# Per-asyncio-task deadline — isolates concurrent requests from each other +_deadline_var: contextvars.ContextVar[Optional[float]] = contextvars.ContextVar( + "sec_http_deadline", default=None +) + logger = logging.getLogger(__name__) @@ -87,7 +93,6 @@ class SECHttpClient: except Exception: pass self._user_agent = f"{user_agent_name} ({settings.SEC_EMAIL})" - self._deadline: Optional[float] = None self._session: Optional[aiohttp.ClientSession] = None async def _get_session(self) -> aiohttp.ClientSession: @@ -110,28 +115,30 @@ class SECHttpClient: # ------------------------------------------------------------------ def set_deadline(self, seconds_from_now: float) -> None: - self._deadline = _time.monotonic() + seconds_from_now + _deadline_var.set(_time.monotonic() + seconds_from_now) def clear_deadline(self) -> None: - self._deadline = None + _deadline_var.set(None) @property def deadline(self) -> Optional[float]: - return self._deadline + return _deadline_var.get() @deadline.setter def deadline(self, value: Optional[float]) -> None: - self._deadline = value + _deadline_var.set(value) def remaining_time(self) -> Optional[float]: - if self._deadline is None: + dl = _deadline_var.get() + if dl is None: return None - return max(0.0, self._deadline - _time.monotonic()) + return max(0.0, dl - _time.monotonic()) def is_deadline_exceeded(self) -> bool: - if self._deadline is None: + dl = _deadline_var.get() + if dl is None: return False - return _time.monotonic() >= self._deadline + return _time.monotonic() >= dl # ------------------------------------------------------------------ # Cache helpers @@ -188,11 +195,12 @@ class SECHttpClient: last_exc = None for _i in range(attempts): now = _time.monotonic() - if self._deadline is not None and now >= self._deadline: + dl = _deadline_var.get() + if dl is not None and now >= dl: break req_timeout = self.http_timeout - if self._deadline is not None: - remaining = max(0.0, self._deadline - now) + if dl is not None: + remaining = max(0.0, dl - now) if remaining < 0.25: break req_timeout = aiohttp.ClientTimeout( @@ -210,8 +218,9 @@ class SECHttpClient: if retry_after and retry_after.isdigit() else backoff ) - if self._deadline is not None: - remaining = max(0.0, self._deadline - _time.monotonic()) + dl = _deadline_var.get() + if dl is not None: + remaining = max(0.0, dl - _time.monotonic()) delay = min(delay, max(0.0, remaining - 0.05)) await asyncio.sleep( max(0.0, delay) @@ -221,8 +230,9 @@ class SECHttpClient: continue if 500 <= resp.status < 600: delay = backoff - if self._deadline is not None: - remaining = max(0.0, self._deadline - _time.monotonic()) + dl = _deadline_var.get() + if dl is not None: + remaining = max(0.0, dl - _time.monotonic()) delay = min(delay, max(0.0, remaining - 0.05)) await asyncio.sleep( max(0.0, delay) @@ -243,8 +253,9 @@ class SECHttpClient: except Exception as e: last_exc = e delay = backoff - if self._deadline is not None: - remaining = max(0.0, self._deadline - _time.monotonic()) + dl = _deadline_var.get() + if dl is not None: + remaining = max(0.0, dl - _time.monotonic()) delay = min(delay, max(0.0, remaining - 0.05)) await asyncio.sleep( max(0.0, delay) @@ -283,11 +294,12 @@ class SECHttpClient: last_exc = None for _i in range(attempts): now = _time.monotonic() - if self._deadline is not None and now >= self._deadline: + dl = _deadline_var.get() + if dl is not None and now >= dl: break req_timeout = self.http_timeout - if self._deadline is not None: - remaining = max(0.0, self._deadline - now) + if dl is not None: + remaining = max(0.0, dl - now) if remaining < 0.25: break req_timeout = aiohttp.ClientTimeout( @@ -305,8 +317,9 @@ class SECHttpClient: if retry_after and retry_after.isdigit() else backoff ) - if self._deadline is not None: - remaining = max(0.0, self._deadline - _time.monotonic()) + dl = _deadline_var.get() + if dl is not None: + remaining = max(0.0, dl - _time.monotonic()) delay = min(delay, max(0.0, remaining - 0.05)) await asyncio.sleep( max(0.0, delay) @@ -316,8 +329,9 @@ class SECHttpClient: continue if 500 <= resp.status < 600: delay = backoff - if self._deadline is not None: - remaining = max(0.0, self._deadline - _time.monotonic()) + dl = _deadline_var.get() + if dl is not None: + remaining = max(0.0, dl - _time.monotonic()) delay = min(delay, max(0.0, remaining - 0.05)) await asyncio.sleep( max(0.0, delay) @@ -330,8 +344,9 @@ class SECHttpClient: if _is_sec_block_page(text): last_exc = RuntimeError("SEC_BLOCKED") delay = backoff * 2.0 - if self._deadline is not None: - remaining = max(0.0, self._deadline - _time.monotonic()) + dl = _deadline_var.get() + if dl is not None: + remaining = max(0.0, dl - _time.monotonic()) delay = min(delay, max(0.0, remaining - 0.05)) await asyncio.sleep( max(0.0, delay) @@ -350,8 +365,9 @@ class SECHttpClient: except Exception as e: last_exc = e delay = backoff - if self._deadline is not None: - remaining = max(0.0, self._deadline - _time.monotonic()) + dl = _deadline_var.get() + if dl is not None: + remaining = max(0.0, dl - _time.monotonic()) delay = min(delay, max(0.0, remaining - 0.05)) await asyncio.sleep( max(0.0, delay)