diff --git a/tools/gitea/gitea_common.py b/tools/gitea/gitea_common.py index 7147bbf..63c0031 100644 --- a/tools/gitea/gitea_common.py +++ b/tools/gitea/gitea_common.py @@ -4,14 +4,13 @@ from __future__ import annotations import dataclasses +import http.client import json import os import pathlib import re import subprocess -import urllib.error import urllib.parse -import urllib.request from typing import Any @@ -90,9 +89,23 @@ def validated_http_base_url(value: str) -> str: class GiteaClient: - def __init__(self, target: RepoTarget, token: str | None) -> None: + def __init__(self, target: RepoTarget, token: str | None, *, timeout: float = 30.0) -> None: self.target = target self.token = token + self.timeout = timeout + self._api_url = urllib.parse.urlparse(target.api_base) + self._connection: http.client.HTTPConnection | None = None + + def __enter__(self) -> GiteaClient: + return self + + def __exit__(self, *_exc_info: object) -> None: + self.close() + + def close(self) -> None: + if self._connection is not None: + self._connection.close() + self._connection = None def request_json( self, @@ -102,14 +115,13 @@ class GiteaClient: body: dict[str, Any] | None = None, query: dict[str, Any] | None = None, ) -> Any: - url = f"{self.target.api_base}{path}" - if query: - url = f"{url}?{urllib.parse.urlencode(query, doseq=True)}" + request_target = self._request_target(path, query) data = None headers = { "Accept": "application/json", "User-Agent": "codex-gitea-maintenance", + "Connection": "keep-alive", } if self.token: headers["Authorization"] = f"token {self.token}" @@ -117,18 +129,71 @@ class GiteaClient: data = json.dumps(body).encode("utf-8") headers["Content-Type"] = "application/json" - request = urllib.request.Request(url, data=data, headers=headers, method=method) + method = method.upper() + safe_to_retry = method in {"GET", "HEAD", "OPTIONS"} + last_error: BaseException | None = None + for attempt in range(2 if safe_to_retry else 1): + try: + status, _reason, payload = self._request_once(method, request_target, data, headers) + break + except (OSError, http.client.HTTPException) as exc: + self.close() + last_error = exc + if attempt == 0 and safe_to_retry: + continue + raise GiteaError(f"{method} {self._display_url(request_target)} failed: {exc}") from exc + else: + raise GiteaError(f"{method} {self._display_url(request_target)} failed: {last_error}") + + if status >= 400: + message = payload.decode("utf-8", errors="replace") + raise GiteaError(f"{method} {self._display_url(request_target)} failed with HTTP {status}: {message}") + if not payload: + return None + return json.loads(payload.decode("utf-8")) + + def _connection_for_request(self) -> http.client.HTTPConnection: + if self._connection is not None: + return self._connection + host = self._api_url.hostname + if not host: + raise GiteaError("Gitea API URL is missing a hostname.") + if self._api_url.scheme == "https": + self._connection = http.client.HTTPSConnection(host, self._api_url.port, timeout=self.timeout) + elif self._api_url.scheme == "http": + self._connection = http.client.HTTPConnection(host, self._api_url.port, timeout=self.timeout) + else: + raise GiteaError("Gitea API URL must use http:// or https://.") + return self._connection + + def _request_once( + self, + method: str, + request_target: str, + data: bytes | None, + headers: dict[str, str], + ) -> tuple[int, str, bytes]: + connection = self._connection_for_request() + connection.request(method, request_target, body=data, headers=headers) + response = connection.getresponse() try: - with urllib.request.urlopen(request, timeout=30) as response: # noqa: S310 - validated Gitea http(s) API URL. # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected - payload = response.read().decode("utf-8") - if not payload: - return None - return json.loads(payload) - except urllib.error.HTTPError as exc: - message = exc.read().decode("utf-8", errors="replace") - raise GiteaError(f"{method} {url} failed with HTTP {exc.code}: {message}") from exc - except urllib.error.URLError as exc: - raise GiteaError(f"{method} {url} failed: {exc.reason}") from exc + payload = response.read() + return response.status, response.reason, payload + finally: + if response.will_close: + self.close() + + def _request_target(self, path: str, query: dict[str, Any] | None) -> str: + base_path = self._api_url.path.rstrip("/") + suffix = path if path.startswith("/") else f"/{path}" + request_target = f"{base_path}{suffix}" + if query: + request_target = f"{request_target}?{urllib.parse.urlencode(query, doseq=True)}" + return request_target + + def _display_url(self, request_target: str) -> str: + netloc = self._api_url.netloc + return urllib.parse.urlunparse((self._api_url.scheme, netloc, request_target, "", "", "")) def paginate( self,