krz/hutch-stats

 1from __future__ import annotations
 2
 3import logging
 4import time
 5from typing import Any
 6
 7import httpx
 8
 9
10logger = logging.getLogger(__name__)
11
12
13class SourceHutClientError(RuntimeError):
14    """Raised when a SourceHut GraphQL request fails."""
15
16
17def _graphql_error_summary(errors: Any) -> str:
18    if not isinstance(errors, list):
19        return "unexpected error payload"
20    return f"{len(errors)} GraphQL error(s)"
21
22
23class SourceHutGraphQLClient:
24    def __init__(
25        self,
26        endpoint: str,
27        token: str,
28        *,
29        timeout: float = 15.0,
30        max_retries: int = 2,
31        request_delay: float = 0.5,
32        transport: httpx.BaseTransport | None = None,
33    ) -> None:
34        self.endpoint = endpoint
35        self.timeout = timeout
36        self.max_retries = max_retries
37        self.request_delay = request_delay
38        headers = {
39            "Authorization": f"Bearer {token}",
40            "Content-Type": "application/json",
41        }
42        self._client = httpx.Client(headers=headers, timeout=timeout, transport=transport)
43
44    def execute(self, query: str, variables: dict[str, Any] | None = None) -> dict[str, Any]:
45        payload = {"query": query, "variables": variables or {}}
46        attempts = self.max_retries + 1
47
48        for attempt in range(attempts):
49            if attempt > 0:
50                time.sleep(2 ** (attempt - 1))
51            elif self.request_delay > 0:
52                time.sleep(self.request_delay)
53
54            try:
55                response = self._client.post(self.endpoint, json=payload)
56                response.raise_for_status()
57                body = response.json()
58            except httpx.HTTPStatusError as exc:
59                logger.warning(
60                    "SourceHut HTTP failure from %s on attempt %s/%s: status=%s",
61                    self.endpoint,
62                    attempt + 1,
63                    attempts,
64                    exc.response.status_code,
65                )
66                if exc.response.status_code >= 500 and attempt < attempts - 1:
67                    continue
68                raise SourceHutClientError(f"HTTP error from SourceHut: {exc.response.status_code}") from exc
69            except httpx.HTTPError as exc:
70                logger.warning(
71                    "SourceHut network failure from %s on attempt %s/%s",
72                    self.endpoint,
73                    attempt + 1,
74                    attempts,
75                )
76                if attempt < attempts - 1:
77                    continue
78                raise SourceHutClientError("Network error while contacting SourceHut") from exc
79
80            if "errors" in body:
81                logger.warning(
82                    "SourceHut GraphQL failure from %s: %s",
83                    self.endpoint,
84                    _graphql_error_summary(body["errors"]),
85                )
86                raise SourceHutClientError(
87                    f"GraphQL errors returned by SourceHut: {_graphql_error_summary(body['errors'])}"
88                )
89            return body.get("data", {})
90
91        raise SourceHutClientError("SourceHut request exhausted retries")
92
93    def close(self) -> None:
94        self._client.close()