Reuse Gitea API connections in maintenance scripts

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,