yirugi
/
stock-oracle
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(sec): _deadline 레이스 컨디션 수정, bulk exhibit 동시성 제한 추가

Browse Source 
- SECHttpClient._deadline(인스턴스 변수) → contextvars.ContextVar로 교체
  asyncio task별 독립 데드라인으로 싱글턴 공유로 인한 레이스 컨디션 해결
- bulk exhibit에 Semaphore(4) + 전체 300s 타임아웃 추가
  동시 50개 코루틴이 Semaphore(2)를 무제한 점유하던 문제 해결

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
main
I Luk Kim 5 months ago
parent f17644e5e8
commit 89113d45ab
2 changed files with 55 additions and 30 deletions
Show Stats Download Patch File Download Diff File

11
app/api/v1/endpoints/filings.py
Unescape Escape View File

@ -314,7 +314,16 @@ async def get_exhibit_bulk(
error=str(e), 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] = [] results: List[BulkExhibitItem] = []
for r in raw: for r in raw:
if isinstance(r, Exception): if isinstance(r, Exception):

74
app/services/sec_http_client.py
Unescape Escape View File

@ -6,6 +6,7 @@ Extracted from etf_holdings_fetcher.py to be reused by all SEC-related services.
import time as _time import time as _time
import asyncio import asyncio
import contextvars
import aiohttp import aiohttp
import hashlib import hashlib
import json import json
@ -16,6 +17,11 @@ from typing import Dict, Optional
from app.core.config import settings 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__) logger = logging.getLogger(__name__)
@ -87,7 +93,6 @@ class SECHttpClient:
except Exception: except Exception:
pass pass
self._user_agent = f"{user_agent_name} ({settings.SEC_EMAIL})" self._user_agent = f"{user_agent_name} ({settings.SEC_EMAIL})"
self._deadline: Optional[float] = None
self._session: Optional[aiohttp.ClientSession] = None self._session: Optional[aiohttp.ClientSession] = None
async def _get_session(self) -> aiohttp.ClientSession: async def _get_session(self) -> aiohttp.ClientSession:
@ -110,28 +115,30 @@ class SECHttpClient:
# ------------------------------------------------------------------ # ------------------------------------------------------------------
def set_deadline(self, seconds_from_now: float) -> None: 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: def clear_deadline(self) -> None:
self._deadline = None _deadline_var.set(None)
@property @property
def deadline(self) -> Optional[float]: def deadline(self) -> Optional[float]:
return self._deadline return _deadline_var.get()
@deadline.setter @deadline.setter
def deadline(self, value: Optional[float]) -> None: def deadline(self, value: Optional[float]) -> None:
self._deadline = value _deadline_var.set(value)
def remaining_time(self) -> Optional[float]: def remaining_time(self) -> Optional[float]:
if self._deadline is None: dl = _deadline_var.get()
if dl is None:
return None return None
return max(0.0, self._deadline - _time.monotonic()) return max(0.0, dl - _time.monotonic())
def is_deadline_exceeded(self) -> bool: def is_deadline_exceeded(self) -> bool:
if self._deadline is None: dl = _deadline_var.get()
if dl is None:
return False return False
return _time.monotonic() >= self._deadline return _time.monotonic() >= dl
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# Cache helpers # Cache helpers
@ -188,11 +195,12 @@ class SECHttpClient:
last_exc = None last_exc = None
for _i in range(attempts): for _i in range(attempts):
now = _time.monotonic() 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 break
req_timeout = self.http_timeout req_timeout = self.http_timeout
if self._deadline is not None: if dl is not None:
remaining = max(0.0, self._deadline - now) remaining = max(0.0, dl - now)
if remaining < 0.25: if remaining < 0.25:
break break
req_timeout = aiohttp.ClientTimeout( req_timeout = aiohttp.ClientTimeout(
@ -210,8 +218,9 @@ class SECHttpClient:
if retry_after and retry_after.isdigit() if retry_after and retry_after.isdigit()
else backoff else backoff
) )
if self._deadline is not None: dl = _deadline_var.get()
remaining = max(0.0, self._deadline - _time.monotonic()) if dl is not None:
remaining = max(0.0, dl - _time.monotonic())
delay = min(delay, max(0.0, remaining - 0.05)) delay = min(delay, max(0.0, remaining - 0.05))
await asyncio.sleep( await asyncio.sleep(
max(0.0, delay) max(0.0, delay)
@ -221,8 +230,9 @@ class SECHttpClient:
continue continue
if 500 <= resp.status < 600: if 500 <= resp.status < 600:
delay = backoff delay = backoff
if self._deadline is not None: dl = _deadline_var.get()
remaining = max(0.0, self._deadline - _time.monotonic()) if dl is not None:
remaining = max(0.0, dl - _time.monotonic())
delay = min(delay, max(0.0, remaining - 0.05)) delay = min(delay, max(0.0, remaining - 0.05))
await asyncio.sleep( await asyncio.sleep(
max(0.0, delay) max(0.0, delay)
@ -243,8 +253,9 @@ class SECHttpClient:
except Exception as e: except Exception as e:
last_exc = e last_exc = e
delay = backoff delay = backoff
if self._deadline is not None: dl = _deadline_var.get()
remaining = max(0.0, self._deadline - _time.monotonic()) if dl is not None:
remaining = max(0.0, dl - _time.monotonic())
delay = min(delay, max(0.0, remaining - 0.05)) delay = min(delay, max(0.0, remaining - 0.05))
await asyncio.sleep( await asyncio.sleep(
max(0.0, delay) max(0.0, delay)
@ -283,11 +294,12 @@ class SECHttpClient:
last_exc = None last_exc = None
for _i in range(attempts): for _i in range(attempts):
now = _time.monotonic() 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 break
req_timeout = self.http_timeout req_timeout = self.http_timeout
if self._deadline is not None: if dl is not None:
remaining = max(0.0, self._deadline - now) remaining = max(0.0, dl - now)
if remaining < 0.25: if remaining < 0.25:
break break
req_timeout = aiohttp.ClientTimeout( req_timeout = aiohttp.ClientTimeout(
@ -305,8 +317,9 @@ class SECHttpClient:
if retry_after and retry_after.isdigit() if retry_after and retry_after.isdigit()
else backoff else backoff
) )
if self._deadline is not None: dl = _deadline_var.get()
remaining = max(0.0, self._deadline - _time.monotonic()) if dl is not None:
remaining = max(0.0, dl - _time.monotonic())
delay = min(delay, max(0.0, remaining - 0.05)) delay = min(delay, max(0.0, remaining - 0.05))
await asyncio.sleep( await asyncio.sleep(
max(0.0, delay) max(0.0, delay)
@ -316,8 +329,9 @@ class SECHttpClient:
continue continue
if 500 <= resp.status < 600: if 500 <= resp.status < 600:
delay = backoff delay = backoff
if self._deadline is not None: dl = _deadline_var.get()
remaining = max(0.0, self._deadline - _time.monotonic()) if dl is not None:
remaining = max(0.0, dl - _time.monotonic())
delay = min(delay, max(0.0, remaining - 0.05)) delay = min(delay, max(0.0, remaining - 0.05))
await asyncio.sleep( await asyncio.sleep(
max(0.0, delay) max(0.0, delay)
@ -330,8 +344,9 @@ class SECHttpClient:
if _is_sec_block_page(text): if _is_sec_block_page(text):
last_exc = RuntimeError("SEC_BLOCKED") last_exc = RuntimeError("SEC_BLOCKED")
delay = backoff * 2.0 delay = backoff * 2.0
if self._deadline is not None: dl = _deadline_var.get()
remaining = max(0.0, self._deadline - _time.monotonic()) if dl is not None:
remaining = max(0.0, dl - _time.monotonic())
delay = min(delay, max(0.0, remaining - 0.05)) delay = min(delay, max(0.0, remaining - 0.05))
await asyncio.sleep( await asyncio.sleep(
max(0.0, delay) max(0.0, delay)
@ -350,8 +365,9 @@ class SECHttpClient:
except Exception as e: except Exception as e:
last_exc = e last_exc = e
delay = backoff delay = backoff
if self._deadline is not None: dl = _deadline_var.get()
remaining = max(0.0, self._deadline - _time.monotonic()) if dl is not None:
remaining = max(0.0, dl - _time.monotonic())
delay = min(delay, max(0.0, remaining - 0.05)) delay = min(delay, max(0.0, remaining - 0.05))
await asyncio.sleep( await asyncio.sleep(
max(0.0, delay) max(0.0, delay)

Write Preview
Loading…
Cancel
Save