fix(emsal): add per-IP rate limiting to prevent spurious empty results
UYAP Emsal (emsal.uyap.gov.tr) rate-limits per source IP, returning HTTP 429 (HTML error page, no Retry-After) after a small burst of rapid requests. With no client-side throttling, sequential searches would fail after the first few — making results appear term-dependent (always the same later queries "returning 0") when the cause was purely request order and rate. On the shared-egress-IP production deployment this was hit constantly. Add the same token-bucket + 429 back-pressure pattern already used by the Bedesten client: requests are spaced ~3.5s apart and the bucket freezes on an actual 429. Configurable via EMSAL_RATE_CAPACITY / EMSAL_RATE_REFILL_S / EMSAL_RATE_MAX_WAIT_S. Verified: seven sequential searches (incl. previously "failing" kıdem, boşanma, kamulaştırma) all return results with no 429s. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
cc055103fe
commit
1b483a6fcf
+107
-3
@@ -6,13 +6,15 @@ import httpx
|
||||
from typing import Dict, Any, List, Optional
|
||||
import logging
|
||||
import html
|
||||
import os
|
||||
import re
|
||||
import io
|
||||
import time
|
||||
from markitdown import MarkItDown
|
||||
|
||||
from .models import (
|
||||
EmsalSearchRequest,
|
||||
EmsalDetailedSearchRequestData,
|
||||
EmsalDetailedSearchRequestData,
|
||||
EmsalApiResponse,
|
||||
EmsalDocumentMarkdown
|
||||
)
|
||||
@@ -21,12 +23,87 @@ logger = logging.getLogger(__name__)
|
||||
if not logger.hasHandlers():
|
||||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
||||
|
||||
|
||||
class EmsalRateLimited(Exception):
|
||||
"""Raised when the local rate-limit bucket would block longer than allowed.
|
||||
|
||||
Carries the suggested retry-after (seconds) so callers can surface a
|
||||
structured 429-style response instead of silently blocking the
|
||||
event-loop slot for the full bucket-pause window.
|
||||
"""
|
||||
|
||||
def __init__(self, retry_after: float) -> None:
|
||||
self.retry_after = retry_after
|
||||
super().__init__(f"local bucket would block {retry_after:.1f}s")
|
||||
|
||||
|
||||
class _TokenBucket:
|
||||
"""Asyncio token bucket with explicit back-pressure.
|
||||
|
||||
The UYAP Emsal endpoint (emsal.uyap.gov.tr) rate-limits per source IP and
|
||||
returns HTTP 429 (an HTML error page, no Retry-After header) after a small
|
||||
burst of rapid requests. On the shared-egress-IP production deployment this
|
||||
is hit constantly, making unrelated searches appear to "return 0 results"
|
||||
depending only on request order. This bucket spaces requests to a safe rate
|
||||
and freezes on an actual 429 via ``penalize_until``.
|
||||
"""
|
||||
|
||||
def __init__(self, capacity: int, refill_per_s: float) -> None:
|
||||
self.capacity = float(capacity)
|
||||
self.refill_per_s = float(refill_per_s)
|
||||
self._tokens = float(capacity)
|
||||
self._last = time.monotonic()
|
||||
self._not_before = 0.0
|
||||
self._lock = asyncio.Lock()
|
||||
|
||||
async def acquire(self, max_wait: Optional[float] = None) -> None:
|
||||
"""Acquire one token. If ``max_wait`` is set and the next wait would
|
||||
exceed it, raise :class:`EmsalRateLimited` immediately instead of
|
||||
sleeping — keeps a single rate-limited request from holding the
|
||||
worker-slot for the full bucket-pause window."""
|
||||
deadline = (time.monotonic() + max_wait) if max_wait is not None else None
|
||||
while True:
|
||||
async with self._lock:
|
||||
now = time.monotonic()
|
||||
if now < self._not_before:
|
||||
wait_s = self._not_before - now
|
||||
else:
|
||||
self._tokens = min(
|
||||
self.capacity,
|
||||
self._tokens + (now - self._last) * self.refill_per_s,
|
||||
)
|
||||
self._last = now
|
||||
if self._tokens >= 1.0:
|
||||
self._tokens -= 1.0
|
||||
return
|
||||
wait_s = (1.0 - self._tokens) / self.refill_per_s
|
||||
if deadline is not None:
|
||||
remaining = deadline - time.monotonic()
|
||||
if wait_s > remaining:
|
||||
raise EmsalRateLimited(retry_after=wait_s)
|
||||
await asyncio.sleep(wait_s)
|
||||
|
||||
def penalize_until(self, monotonic_deadline: float) -> None:
|
||||
"""Pause the bucket until ``monotonic_deadline`` (drains tokens)."""
|
||||
self._not_before = max(self._not_before, monotonic_deadline)
|
||||
self._tokens = 0.0
|
||||
self._last = time.monotonic()
|
||||
|
||||
class EmsalApiClient:
|
||||
"""API Client for Emsal (UYAP Precedent Decision) search system."""
|
||||
BASE_URL = "https://emsal.uyap.gov.tr"
|
||||
DETAILED_SEARCH_ENDPOINT = "/aramadetaylist"
|
||||
DETAILED_SEARCH_ENDPOINT = "/aramadetaylist"
|
||||
DOCUMENT_ENDPOINT = "/getDokuman"
|
||||
|
||||
# UYAP Emsal rate-limits per source IP. Defaults mirror the sibling
|
||||
# Bedesten client (conservative: no burst, ~3.5s spacing). Override via env:
|
||||
# EMSAL_RATE_CAPACITY (default 1)
|
||||
# EMSAL_RATE_REFILL_S (default 3.5; seconds per token)
|
||||
# EMSAL_RATE_MAX_WAIT_S (default 8.0; max local wait before a structured 429)
|
||||
_DEFAULT_CAPACITY = int(os.getenv("EMSAL_RATE_CAPACITY", "1"))
|
||||
_DEFAULT_REFILL_S = float(os.getenv("EMSAL_RATE_REFILL_S", "3.5"))
|
||||
_DEFAULT_MAX_WAIT_S = float(os.getenv("EMSAL_RATE_MAX_WAIT_S", "8.0"))
|
||||
|
||||
def __init__(self, request_timeout: float = 30.0):
|
||||
self.http_client = httpx.AsyncClient(
|
||||
base_url=self.BASE_URL,
|
||||
@@ -38,6 +115,27 @@ class EmsalApiClient:
|
||||
timeout=request_timeout,
|
||||
verify=False # As per user's original FastAPI code
|
||||
)
|
||||
self._bucket = _TokenBucket(
|
||||
capacity=self._DEFAULT_CAPACITY,
|
||||
refill_per_s=1.0 / self._DEFAULT_REFILL_S,
|
||||
)
|
||||
|
||||
def _handle_429(self, response: httpx.Response, op: str) -> None:
|
||||
"""Apply back-pressure to the shared bucket based on Retry-After.
|
||||
|
||||
Emsal returns 429 as an HTML error page with no Retry-After header, so
|
||||
the 30s fallback almost always applies."""
|
||||
retry_after_raw = response.headers.get("Retry-After", "")
|
||||
try:
|
||||
retry_after = float(retry_after_raw)
|
||||
except (TypeError, ValueError):
|
||||
retry_after = 30.0
|
||||
# Cap penalty so a hostile/buggy server can't freeze us indefinitely.
|
||||
retry_after = max(1.0, min(retry_after, 60.0))
|
||||
self._bucket.penalize_until(time.monotonic() + retry_after + 0.5)
|
||||
logger.warning(
|
||||
f"EmsalApiClient: 429 on {op}; bucket paused {retry_after + 0.5:.1f}s"
|
||||
)
|
||||
|
||||
async def search_detailed_decisions(
|
||||
self,
|
||||
@@ -76,7 +174,10 @@ class EmsalApiClient:
|
||||
async def _execute_api_search(self, endpoint: str, payload: Dict) -> EmsalApiResponse:
|
||||
"""Helper method to execute search POST request and process response for Emsal."""
|
||||
try:
|
||||
await self._bucket.acquire(max_wait=self._DEFAULT_MAX_WAIT_S)
|
||||
response = await self.http_client.post(endpoint, json=payload)
|
||||
if response.status_code == 429:
|
||||
self._handle_429(response, "search")
|
||||
response.raise_for_status()
|
||||
response_json_data = response.json()
|
||||
logger.debug(f"EmsalApiClient: Raw API response from {endpoint}: {response_json_data}")
|
||||
@@ -143,9 +244,12 @@ class EmsalApiClient:
|
||||
logger.info(f"EmsalApiClient: Fetching Emsal document for Markdown (ID: {id}) from {source_url}")
|
||||
|
||||
try:
|
||||
await self._bucket.acquire(max_wait=self._DEFAULT_MAX_WAIT_S)
|
||||
response = await self.http_client.get(document_api_url)
|
||||
if response.status_code == 429:
|
||||
self._handle_429(response, f"document {id}")
|
||||
response.raise_for_status()
|
||||
|
||||
|
||||
# Emsal /getDokuman returns JSON with HTML in 'data' field (confirmed by user example)
|
||||
response_json = response.json()
|
||||
html_content_from_api = response_json.get("data")
|
||||
|
||||
Reference in New Issue
Block a user