Compare commits

...

1 Commits

Author SHA1 Message Date
64da98c2bf Add exponential backoff retry logic to GiteaClient 2026-04-06 17:44:07 +00:00

View File

@@ -19,6 +19,7 @@ import os
import urllib.request
import urllib.error
import urllib.parse
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
@@ -211,37 +212,53 @@ class GiteaClient:
# -- HTTP layer ----------------------------------------------------------
def _request(
self,
method: str,
path: str,
data: Optional[dict] = None,
params: Optional[dict] = None,
retries: int = 3,
backoff: float = 1.5,
) -> Any:
"""Make an authenticated API request. Returns parsed JSON."""
"""Make an authenticated API request with exponential backoff retries."""
url = f"{self.api}{path}"
if params:
url += "?" + urllib.parse.urlencode(params)
body = json.dumps(data).encode() if data else None
req = urllib.request.Request(url, data=body, method=method)
req.add_header("Authorization", f"token {self.token}")
req.add_header("Content-Type", "application/json")
req.add_header("Accept", "application/json")
for attempt in range(retries):
req = urllib.request.Request(url, data=body, method=method)
req.add_header("Authorization", f"token {self.token}")
req.add_header("Content-Type", "application/json")
req.add_header("Accept", "application/json")
try:
with urllib.request.urlopen(req, timeout=30) as resp:
raw = resp.read().decode()
if not raw:
return {}
return json.loads(raw)
except urllib.error.HTTPError as e:
body_text = ""
try:
body_text = e.read().decode()
except Exception:
pass
raise GiteaError(e.code, body_text, url) from e
with urllib.request.urlopen(req, timeout=30) as resp:
raw = resp.read().decode()
if not raw:
return {}
return json.loads(raw)
except urllib.error.HTTPError as e:
# Don't retry client errors (4xx) except 429
if 400 <= e.code < 500 and e.code != 429:
body_text = ""
try:
body_text = e.read().decode()
except Exception:
pass
raise GiteaError(e.code, body_text, url) from e
if attempt == retries - 1:
raise GiteaError(e.code, str(e), url) from e
time.sleep(backoff ** attempt)
except (urllib.error.URLError, TimeoutError) as e:
if attempt == retries - 1:
raise GiteaError(500, str(e), url) from e
time.sleep(backoff ** attempt)
def _get(self, path: str, **params) -> Any:
# Filter out None values