48 lines
1.8 KiB
Python
48 lines
1.8 KiB
Python
"""Shared, bounded provider liveness probes."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Iterable
|
|
|
|
import requests
|
|
|
|
|
|
def probe_provider(
|
|
base_url: str,
|
|
checks: Iterable[tuple[str, str | None]],
|
|
*,
|
|
headers: dict[str, str] | None = None,
|
|
timeout: float = 3.0,
|
|
) -> dict[str, Any]:
|
|
"""Try provider endpoints in order with normalized, body-free results."""
|
|
base = str(base_url or '').rstrip('/')
|
|
limit = max(0.5, float(timeout))
|
|
last_error = 'No response'
|
|
for path, json_key in checks:
|
|
endpoint = path or '/'
|
|
try:
|
|
response = requests.get(base + path, timeout=limit, headers=headers or {}, allow_redirects=True)
|
|
if response.status_code >= 500:
|
|
last_error = f'HTTP {response.status_code} on {endpoint}'
|
|
continue
|
|
if response.status_code >= 400:
|
|
last_error = f'HTTP {response.status_code} on {endpoint}'
|
|
continue
|
|
payload = None
|
|
if json_key:
|
|
try:
|
|
payload = response.json()
|
|
except Exception:
|
|
last_error = f'{endpoint} returned non-JSON (HTTP {response.status_code})'
|
|
continue
|
|
if not isinstance(payload, dict) or json_key not in payload:
|
|
last_error = f"Unexpected response from {endpoint} (missing '{json_key}')"
|
|
continue
|
|
return {'ok': True, 'status': response.status_code, 'endpoint': endpoint, 'payload': payload}
|
|
except requests.exceptions.ConnectionError:
|
|
last_error = 'Connection refused'
|
|
except requests.exceptions.Timeout:
|
|
last_error = f'Timeout ({limit:g} s)'
|
|
except Exception as exc:
|
|
last_error = str(exc)
|
|
return {'ok': False, 'error': last_error}
|