309 lines
11 KiB
Python
309 lines
11 KiB
Python
from __future__ import annotations
|
||
|
||
import logging
|
||
import time
|
||
from collections.abc import Callable, Mapping
|
||
from dataclasses import dataclass, field
|
||
from typing import Any, Protocol
|
||
|
||
import requests
|
||
from pydantic import SecretStr
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
API_KEY_HEADER = "X-Rshb-Api-Key"
|
||
|
||
# Идентификатор запроса в Gravitee — по нему ищутся логи шлюза (D18). Это
|
||
# единственный заголовок, который сохраняем: в нём нет секретов.
|
||
TRANSACTION_ID_HEADER = "X-Gravitee-Transaction-Id"
|
||
|
||
# Ретраим только то, что может пройти само: перегрузка шлюза/источника и сеть.
|
||
RETRY_STATUSES = frozenset({429, 502, 503, 504})
|
||
|
||
# Так Gravitee отвечает на путь, для которого нет API (опечатка в MOEX_PATH /
|
||
# CBR_PATH), — это ошибка конфигурации, а не «бумага не найдена».
|
||
NO_CONTEXT_PATH_MARKER = "No context-path matches the request URI"
|
||
|
||
_MAX_RETRY_AFTER_SECONDS = 30.0
|
||
_BODY_SNIPPET_LENGTH = 200
|
||
|
||
|
||
class GatewayError(RuntimeError):
|
||
"""Ошибка обращения к внешнему источнику через шлюз Gravitee.
|
||
|
||
Текст исключения — для пользователя (по-русски, без ключей и заголовков).
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
message: str,
|
||
*,
|
||
source: str,
|
||
endpoint: str,
|
||
http_status: int | None = None,
|
||
transaction_id: str | None = None,
|
||
) -> None:
|
||
super().__init__(message)
|
||
self.source = source
|
||
self.endpoint = endpoint
|
||
self.http_status = http_status
|
||
self.transaction_id = transaction_id
|
||
|
||
|
||
class GatewayAuthError(GatewayError):
|
||
"""401/403: ключ не задан, неверен или выдан на другой ресурс."""
|
||
|
||
|
||
class GatewayNotFoundError(GatewayError):
|
||
"""404 «No context-path matches…»: неверный path шлюза (MOEX_PATH/CBR_PATH)."""
|
||
|
||
|
||
class UpstreamNotFoundError(GatewayError):
|
||
"""404 от самого источника (шлюз маршрут нашёл, ресурса у источника нет)."""
|
||
|
||
|
||
class GatewayUnavailableError(GatewayError):
|
||
"""Сеть/таймаут/429/5xx — повторы исчерпаны."""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class RequestRecord:
|
||
"""Одна логическая операция GET (со всеми повторами) — для external_request_log.
|
||
|
||
Сознательно без тела ответа и без заголовков — кроме transaction_id
|
||
(X-Gravitee-Transaction-Id последней попытки, D18).
|
||
"""
|
||
|
||
source: str
|
||
endpoint: str
|
||
params: dict[str, Any]
|
||
http_status: int | None
|
||
duration_ms: int
|
||
ok: bool
|
||
error: str | None
|
||
transaction_id: str | None = None
|
||
|
||
|
||
def transaction_id_of(response: requests.Response | None) -> str | None:
|
||
if response is None:
|
||
return None
|
||
return response.headers.get(TRANSACTION_ID_HEADER) or None
|
||
|
||
|
||
class HttpSession(Protocol):
|
||
def get(self, url: str, **kwargs: Any) -> requests.Response: ...
|
||
|
||
|
||
@dataclass
|
||
class GatewayClient:
|
||
"""Синхронный клиент к шлюзу Gravitee для одного ресурса (MOEX или ЦБ).
|
||
|
||
Синхронный намеренно: Streamlit перезапускает скрипт на каждый клик,
|
||
запросов единицы — async для HTTP ничего не даёт (как в
|
||
../sfm/app/opencloud_access.py).
|
||
|
||
`retries` — число повторов после первой попытки (всего до retries + 1
|
||
запросов). Пауза между попытками — backoff_seconds * 2**n, для 429 —
|
||
Retry-After, если он есть (не больше 30 с).
|
||
"""
|
||
|
||
source: str
|
||
base_url: str
|
||
path_prefix: str
|
||
api_key: SecretStr
|
||
timeout_seconds: float = 15
|
||
retries: int = 3
|
||
backoff_seconds: float = 1.0
|
||
session: HttpSession = field(default_factory=requests.Session)
|
||
sleep: Callable[[float], None] = time.sleep
|
||
records: list[RequestRecord] = field(default_factory=list)
|
||
|
||
def endpoint(self, path: str) -> str:
|
||
"""Путь на шлюзе без хоста — так он пишется в журнал."""
|
||
return f"{self.path_prefix.rstrip('/')}/{path.lstrip('/')}"
|
||
|
||
def drain_records(self) -> list[RequestRecord]:
|
||
"""Забирает накопленные записи журнала (для сохранения в БД)."""
|
||
records, self.records = self.records, []
|
||
return records
|
||
|
||
def get(
|
||
self, path: str, params: Mapping[str, Any] | None = None
|
||
) -> requests.Response:
|
||
endpoint = self.endpoint(path)
|
||
params_dict = dict(params or {})
|
||
started = time.monotonic()
|
||
try:
|
||
response = self._get_with_retries(endpoint, params_dict)
|
||
except GatewayError as exc:
|
||
self._record(
|
||
endpoint,
|
||
params_dict,
|
||
started,
|
||
exc.http_status,
|
||
str(exc),
|
||
exc.transaction_id,
|
||
)
|
||
raise
|
||
self._record(
|
||
endpoint,
|
||
params_dict,
|
||
started,
|
||
response.status_code,
|
||
None,
|
||
transaction_id_of(response),
|
||
)
|
||
return response
|
||
|
||
def _get_with_retries(
|
||
self, endpoint: str, params: dict[str, Any]
|
||
) -> requests.Response:
|
||
key = self.api_key.get_secret_value()
|
||
if not key:
|
||
raise GatewayAuthError(
|
||
f"Не задан ключ шлюза для {self.source.upper()} "
|
||
f"(OPENBAO__SETTINGS__{self.source.upper()}_API_KEY).",
|
||
source=self.source,
|
||
endpoint=endpoint,
|
||
)
|
||
|
||
url = f"{self.base_url.rstrip('/')}{endpoint}"
|
||
attempts = self.retries + 1
|
||
for attempt in range(1, attempts + 1):
|
||
try:
|
||
response = self.session.get(
|
||
url,
|
||
params=params,
|
||
headers={API_KEY_HEADER: key},
|
||
timeout=self.timeout_seconds,
|
||
)
|
||
except (requests.Timeout, requests.ConnectionError) as exc:
|
||
reason = (
|
||
"таймаут" if isinstance(exc, requests.Timeout) else "ошибка сети"
|
||
)
|
||
logger.warning(
|
||
"%s GET %s: %s (попытка %d из %d)",
|
||
self.source,
|
||
endpoint,
|
||
reason,
|
||
attempt,
|
||
attempts,
|
||
)
|
||
if attempt == attempts:
|
||
raise GatewayUnavailableError(
|
||
f"{self.source.upper()} недоступен через шлюз: {reason} "
|
||
f"после {attempts} попыток.",
|
||
source=self.source,
|
||
endpoint=endpoint,
|
||
) from None
|
||
self.sleep(self._backoff(attempt, None))
|
||
continue
|
||
|
||
status = response.status_code
|
||
if status < 400:
|
||
return response
|
||
|
||
if status in RETRY_STATUSES:
|
||
logger.warning(
|
||
"%s GET %s: HTTP %d (попытка %d из %d)",
|
||
self.source,
|
||
endpoint,
|
||
status,
|
||
attempt,
|
||
attempts,
|
||
)
|
||
if attempt == attempts:
|
||
raise GatewayUnavailableError(
|
||
f"{self.source.upper()} временно недоступен: HTTP {status} "
|
||
f"после {attempts} попыток.",
|
||
source=self.source,
|
||
endpoint=endpoint,
|
||
http_status=status,
|
||
transaction_id=transaction_id_of(response),
|
||
)
|
||
self.sleep(self._backoff(attempt, response))
|
||
continue
|
||
|
||
raise self._non_retryable_error(endpoint, response)
|
||
|
||
raise AssertionError("unreachable") # pragma: no cover
|
||
|
||
def _non_retryable_error(
|
||
self, endpoint: str, response: requests.Response
|
||
) -> GatewayError:
|
||
status = response.status_code
|
||
# Тело только для диагностики и коротко; 401 от шлюза — plain text
|
||
# («Unauthorized»), не JSON, поэтому не парсим.
|
||
body = (response.text or "").strip()[:_BODY_SNIPPET_LENGTH]
|
||
source = self.source.upper()
|
||
common = {
|
||
"source": self.source,
|
||
"endpoint": endpoint,
|
||
"http_status": status,
|
||
"transaction_id": transaction_id_of(response),
|
||
}
|
||
|
||
if status in (401, 403):
|
||
return GatewayAuthError(
|
||
f"Шлюз отказал в доступе к {source} (HTTP {status}). Проверьте "
|
||
f"OPENBAO__SETTINGS__{source}_API_KEY — ключи у MOEX и ЦБ разные.",
|
||
**common,
|
||
)
|
||
if status == 404 and NO_CONTEXT_PATH_MARKER in body:
|
||
return GatewayNotFoundError(
|
||
f"Шлюз не знает путь {endpoint} (HTTP 404, «{NO_CONTEXT_PATH_MARKER}»). "
|
||
f"Проверьте OPENBAO__SETTINGS__{source}_PATH.",
|
||
**common,
|
||
)
|
||
if status == 404:
|
||
return UpstreamNotFoundError(
|
||
f"{source}: ресурс не найден (HTTP 404).",
|
||
**common,
|
||
)
|
||
return GatewayError(
|
||
f"{source} вернул ошибку HTTP {status}: {body}",
|
||
**common,
|
||
)
|
||
|
||
def _backoff(self, attempt: int, response: requests.Response | None) -> float:
|
||
if response is not None and response.status_code == 429:
|
||
retry_after = response.headers.get("Retry-After", "")
|
||
if retry_after.isdigit():
|
||
return min(float(retry_after), _MAX_RETRY_AFTER_SECONDS)
|
||
return self.backoff_seconds * 2 ** (attempt - 1)
|
||
|
||
def _record(
|
||
self,
|
||
endpoint: str,
|
||
params: dict[str, Any],
|
||
started: float,
|
||
http_status: int | None,
|
||
error: str | None,
|
||
transaction_id: str | None,
|
||
) -> None:
|
||
duration_ms = int((time.monotonic() - started) * 1000)
|
||
ok = error is None
|
||
self.records.append(
|
||
RequestRecord(
|
||
source=self.source,
|
||
endpoint=endpoint,
|
||
params=params,
|
||
http_status=http_status,
|
||
duration_ms=duration_ms,
|
||
ok=ok,
|
||
error=error,
|
||
transaction_id=transaction_id,
|
||
)
|
||
)
|
||
logger.log(
|
||
logging.INFO if ok else logging.WARNING,
|
||
"%s GET %s params=%s -> %s за %d мс, tx=%s%s",
|
||
self.source,
|
||
endpoint,
|
||
params,
|
||
http_status if http_status is not None else "—",
|
||
duration_ms,
|
||
transaction_id or "—",
|
||
"" if ok else f": {error}",
|
||
)
|