Spark
3e3187d597
- _parallel_map : asyncio.gather borné (16) pour dashboard, /health, /forecast, /triage et le mur — le patron de /focus, généralisé. Avant : la page attendait la somme des allers- retours Gitea de tous les dépôts, en séquentiel. - Client Gitea : les listes (dépôts, issues, jalons, étoiles) passent en stale-while-revalidate — une entrée périmée est servie immédiatement et rafraîchie en tâche de fond. Les mutations vident le cache et bumpent une génération qui désarme les rafraîchissements en vol. - /forecast calcule des lignes de jalons tous projets (écart ETA/échéance) ; /health trie par sévérité et expose la pastille ; le détail santé reçoit la liste des projets.
442 lines
16 KiB
Python
442 lines
16 KiB
Python
"""
|
|
Async Gitea API client for Kanbante.
|
|
|
|
Changes from the 0.1.0 client:
|
|
- Fully asynchronous (httpx.AsyncClient): routes no longer block the
|
|
event loop on Gitea calls.
|
|
- Real pagination: Gitea caps every list response at MAX_RESPONSE_ITEMS
|
|
(50 by default) and silently truncates larger `limit` values, so all
|
|
list methods follow pages until the result is complete.
|
|
- The Authorization header is only sent when a token is configured.
|
|
- The current user's id is resolved from /user (no hardcoded uid or
|
|
username); without a token it falls back to a public search.
|
|
- Write operations used by the two-way sync: create_issue() and
|
|
set_issue_state().
|
|
- Short in-memory cache for list responses (KANBANTE_CACHE_TTL seconds),
|
|
invalidated on every mutation.
|
|
- .env loading targets the project root only (no hardcoded home path).
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import time
|
|
from collections.abc import Awaitable, Callable
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from dotenv import load_dotenv
|
|
|
|
from app.config import get_settings
|
|
|
|
load_dotenv(Path(__file__).resolve().parent.parent / ".env")
|
|
|
|
logger = logging.getLogger("kanbante.gitea")
|
|
|
|
PAGE_SIZE = 50 # Gitea's default MAX_RESPONSE_ITEMS
|
|
MAX_PAGES = 40 # hard stop: 40 * 50 = 2000 items per list
|
|
|
|
|
|
class GiteaClient:
|
|
"""Lightweight async Gitea API client for Kanbante."""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: str | None = None,
|
|
token: str | None = None,
|
|
cache_ttl: float | None = None,
|
|
transport: httpx.AsyncBaseTransport | None = None,
|
|
) -> None:
|
|
settings = get_settings()
|
|
self.base_url = (base_url or settings.gitea_url).rstrip("/")
|
|
self.token = settings.gitea_token if token is None else token
|
|
self.cache_ttl = settings.cache_ttl if cache_ttl is None else cache_ttl
|
|
headers = {"Accept": "application/json"}
|
|
if self.token:
|
|
headers["Authorization"] = f"token {self.token}"
|
|
self.client = httpx.AsyncClient(
|
|
base_url=f"{self.base_url}/api/v1",
|
|
headers=headers,
|
|
timeout=15.0,
|
|
transport=transport,
|
|
)
|
|
self._user: dict | None = None
|
|
self._cache: dict[str, tuple[float, Any]] = {}
|
|
self._refreshing: set[str] = set()
|
|
self._cache_gen = 0
|
|
|
|
async def aclose(self) -> None:
|
|
await self.client.aclose()
|
|
|
|
# ── Low-level helpers ──────────────────────────────────
|
|
|
|
async def _get(self, path: str, **params: Any) -> Any:
|
|
r = await self.client.get(path, params=params)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
async def _get_all_pages(self, path: str, **params: Any) -> list[dict]:
|
|
"""Fetch every page of a list endpoint.
|
|
|
|
Stops on a short page, on X-Total-Count when Gitea provides it,
|
|
or at MAX_PAGES as a safety net.
|
|
"""
|
|
items: list[dict] = []
|
|
page = 1
|
|
while page <= MAX_PAGES:
|
|
r = await self.client.get(
|
|
path, params={**params, "page": page, "limit": PAGE_SIZE}
|
|
)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
if isinstance(data, dict) and "data" in data:
|
|
batch = data["data"]
|
|
elif isinstance(data, list):
|
|
batch = data
|
|
else:
|
|
batch = []
|
|
items.extend(batch)
|
|
|
|
total = r.headers.get("X-Total-Count")
|
|
if total is not None:
|
|
try:
|
|
if len(items) >= int(total):
|
|
break
|
|
except ValueError:
|
|
pass
|
|
if len(batch) < PAGE_SIZE:
|
|
break
|
|
page += 1
|
|
return items
|
|
|
|
def _cache_put(self, key: str, value: Any) -> None:
|
|
if self.cache_ttl > 0:
|
|
self._cache[key] = (time.monotonic(), value)
|
|
|
|
def _cache_clear(self) -> None:
|
|
self._cache.clear()
|
|
# Invalidate in-flight background refreshes: a refresh that
|
|
# started before a mutation must not resurrect stale data.
|
|
self._cache_gen += 1
|
|
|
|
async def _cached_list(
|
|
self, key: str, fetcher: Callable[[], Awaitable[Any]]
|
|
) -> Any:
|
|
"""Stale-while-revalidate cache for list endpoints (0.19.0).
|
|
|
|
Fresh (< TTL): serve the cache. Stale: serve the cached
|
|
value IMMEDIATELY and refresh it in the background (once
|
|
per key) — the user never waits on Gitea for a page when
|
|
any earlier answer exists. Absent: fetch and wait.
|
|
Mutations clear the cache (see _cache_clear), so writes are
|
|
never hidden by a stale entry.
|
|
"""
|
|
if self.cache_ttl <= 0:
|
|
return await fetcher()
|
|
hit = self._cache.get(key)
|
|
if hit is not None:
|
|
if (time.monotonic() - hit[0]) < self.cache_ttl:
|
|
return hit[1]
|
|
if key not in self._refreshing:
|
|
self._refreshing.add(key)
|
|
asyncio.create_task(
|
|
self._refresh_entry(key, fetcher, self._cache_gen)
|
|
)
|
|
return hit[1]
|
|
value = await fetcher()
|
|
self._cache_put(key, value)
|
|
return value
|
|
|
|
async def _refresh_entry(
|
|
self,
|
|
key: str,
|
|
fetcher: Callable[[], Awaitable[Any]],
|
|
generation: int,
|
|
) -> None:
|
|
try:
|
|
value = await fetcher()
|
|
if generation == self._cache_gen:
|
|
self._cache_put(key, value)
|
|
except Exception as exc:
|
|
logger.debug("background refresh failed for %s: %s", key, exc)
|
|
finally:
|
|
self._refreshing.discard(key)
|
|
|
|
def invalidate_cache(self) -> None:
|
|
"""Drop every cached list response (called on webhook events)."""
|
|
self._cache_clear()
|
|
|
|
# ── API ────────────────────────────────────────────────
|
|
|
|
async def ping(self) -> bool:
|
|
"""True if the Gitea instance answers /version."""
|
|
try:
|
|
r = await self.client.get("/version", timeout=5.0)
|
|
return r.status_code == 200
|
|
except Exception:
|
|
return False
|
|
|
|
async def get_user(self) -> dict:
|
|
"""Get authenticated user info (cached on the instance)."""
|
|
if self._user is None:
|
|
self._user = await self._get("/user")
|
|
return self._user
|
|
|
|
async def search_repos(self) -> list[dict]:
|
|
"""All repos visible to the token owner (paginated, SWR-cached)."""
|
|
|
|
async def _fetch() -> list[dict]:
|
|
params: dict[str, Any] = {}
|
|
try:
|
|
user = await self.get_user()
|
|
if user and user.get("id"):
|
|
params["uid"] = user["id"]
|
|
except Exception as exc: # no/invalid token → public search
|
|
logger.debug(
|
|
"get_user failed, falling back to public search: %s", exc
|
|
)
|
|
return await self._get_all_pages("/repos/search", **params)
|
|
|
|
return await self._cached_list("search_repos", _fetch)
|
|
|
|
async def get_repo(self, owner: str, repo: str) -> dict:
|
|
return await self._get(f"/repos/{owner}/{repo}")
|
|
|
|
async def get_user_repos(self, username: str) -> list[dict]:
|
|
"""Get all repos for a user (paginated)."""
|
|
return await self._get_all_pages(f"/users/{username}/repos")
|
|
|
|
async def get_org_repos(self, org: str) -> list[dict]:
|
|
"""Get all repos for an organization (paginated)."""
|
|
return await self._get_all_pages(f"/orgs/{org}/repos")
|
|
|
|
async def get_repo_issues(
|
|
self, owner: str, repo: str, state: str = "all"
|
|
) -> list[dict]:
|
|
"""Get issues for a repo (paginated, cached).
|
|
|
|
Note: Gitea's issues endpoint also returns pull requests; a
|
|
PR carries a non-null "pull_request" object, a plain issue
|
|
carries "pull_request": null — callers filter on truthiness
|
|
(see app.main._is_issue).
|
|
"""
|
|
key = f"issues:{owner}/{repo}:{state}"
|
|
return await self._cached_list(
|
|
key,
|
|
lambda: self._get_all_pages(
|
|
f"/repos/{owner}/{repo}/issues", state=state
|
|
),
|
|
)
|
|
|
|
async def get_all_issues(
|
|
self, repos: list[dict], state: str = "open"
|
|
) -> dict[str, list[dict]]:
|
|
"""Get issues for multiple repos, keyed by full_name."""
|
|
issues_by_repo: dict[str, list[dict]] = {}
|
|
for repo in repos:
|
|
full_name = repo["full_name"]
|
|
owner, name = full_name.split("/", 1)
|
|
try:
|
|
issues_by_repo[full_name] = await self.get_repo_issues(
|
|
owner, name, state=state
|
|
)
|
|
except Exception as exc:
|
|
logger.warning("issues fetch failed for %s: %s", full_name, exc)
|
|
issues_by_repo[full_name] = []
|
|
return issues_by_repo
|
|
|
|
async def get_issue(self, owner: str, repo: str, number: int) -> dict:
|
|
"""One issue with body, labels, assignees, milestone."""
|
|
return await self._get(f"/repos/{owner}/{repo}/issues/{number}")
|
|
|
|
async def get_issue_comments(
|
|
self, owner: str, repo: str, number: int
|
|
) -> list[dict]:
|
|
"""All comments of an issue (paginated)."""
|
|
return await self._get_all_pages(
|
|
f"/repos/{owner}/{repo}/issues/{number}/comments"
|
|
)
|
|
|
|
async def create_comment(
|
|
self, owner: str, repo: str, number: int, body: str
|
|
) -> dict:
|
|
"""Post a comment on an issue and return it."""
|
|
r = await self.client.post(
|
|
f"/repos/{owner}/{repo}/issues/{number}/comments",
|
|
json={"body": body},
|
|
)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
# ── Labels (kanban:* column mirroring) ─────────────────
|
|
|
|
async def get_labels(self, owner: str, repo: str) -> list[dict]:
|
|
return await self._get_all_pages(f"/repos/{owner}/{repo}/labels")
|
|
|
|
async def create_label(
|
|
self, owner: str, repo: str, name: str, color: str
|
|
) -> dict:
|
|
r = await self.client.post(
|
|
f"/repos/{owner}/{repo}/labels",
|
|
json={"name": name, "color": color.lstrip("#")},
|
|
)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
async def ensure_label(
|
|
self, owner: str, repo: str, name: str, color: str
|
|
) -> dict:
|
|
"""Return the repo label with this name, creating it if needed."""
|
|
key = f"{owner}/{repo}"
|
|
cache = self.__dict__.setdefault("_label_cache", {})
|
|
if key not in cache:
|
|
cache[key] = {l["name"]: l for l in await self.get_labels(owner, repo)}
|
|
if name not in cache[key]:
|
|
cache[key][name] = await self.create_label(owner, repo, name, color)
|
|
return cache[key][name]
|
|
|
|
async def add_issue_label(
|
|
self, owner: str, repo: str, number: int, label_id: int
|
|
) -> None:
|
|
r = await self.client.post(
|
|
f"/repos/{owner}/{repo}/issues/{number}/labels",
|
|
json={"labels": [label_id]},
|
|
)
|
|
r.raise_for_status()
|
|
|
|
async def remove_issue_label(
|
|
self, owner: str, repo: str, number: int, label_id: int
|
|
) -> None:
|
|
r = await self.client.delete(
|
|
f"/repos/{owner}/{repo}/issues/{number}/labels/{label_id}"
|
|
)
|
|
r.raise_for_status()
|
|
|
|
async def delete_label(self, owner: str, repo: str, label_id: int) -> None:
|
|
"""Delete a label definition from the repo (all issues lose it)."""
|
|
r = await self.client.delete(f"/repos/{owner}/{repo}/labels/{label_id}")
|
|
r.raise_for_status()
|
|
self.forget_label_cache(owner, repo)
|
|
|
|
def forget_label_cache(self, owner: str, repo: str) -> None:
|
|
cache = self.__dict__.get("_label_cache")
|
|
if cache:
|
|
cache.pop(f"{owner}/{repo}", None)
|
|
|
|
# ── Issue search (global + assigned) ───────────────────
|
|
|
|
async def search_issues(
|
|
self,
|
|
q: str = "",
|
|
state: str = "open",
|
|
assigned_by: str | None = None,
|
|
) -> list[dict]:
|
|
"""Search issues across every repo visible to the token.
|
|
|
|
Uses Gitea's /repos/issues/search (type=issues excludes PRs).
|
|
Results carry their repository in issue["repository"].
|
|
"""
|
|
params: dict[str, Any] = {"state": state, "type": "issues"}
|
|
if q:
|
|
params["q"] = q
|
|
if assigned_by:
|
|
params["assigned_by"] = assigned_by
|
|
return await self._get_all_pages("/repos/issues/search", **params)
|
|
|
|
# ── Milestones ─────────────────────────────────────────
|
|
|
|
async def get_milestones(
|
|
self, owner: str, repo: str, state: str = "open"
|
|
) -> list[dict]:
|
|
key = f"milestones:{owner}/{repo}:{state}"
|
|
return await self._cached_list(
|
|
key,
|
|
lambda: self._get_all_pages(
|
|
f"/repos/{owner}/{repo}/milestones", state=state
|
|
),
|
|
)
|
|
|
|
async def list_starred(self) -> list[dict]:
|
|
"""Repos starred by the token user (issue #20: pinning rides
|
|
on Gitea stars — the forge stays the source of truth)."""
|
|
return await self._cached_list(
|
|
"user:starred", lambda: self._get_all_pages("/user/starred")
|
|
)
|
|
|
|
async def set_starred(self, owner: str, repo: str, starred: bool) -> None:
|
|
"""Star / unstar a repo for the token user."""
|
|
if starred:
|
|
r = await self.client.put(f"/user/starred/{owner}/{repo}")
|
|
else:
|
|
r = await self.client.delete(f"/user/starred/{owner}/{repo}")
|
|
r.raise_for_status()
|
|
self._cache_clear()
|
|
|
|
async def create_issue(
|
|
self, owner: str, repo: str, title: str, body: str = ""
|
|
) -> dict:
|
|
"""Create an issue in Gitea and return it."""
|
|
r = await self.client.post(
|
|
f"/repos/{owner}/{repo}/issues", json={"title": title, "body": body}
|
|
)
|
|
r.raise_for_status()
|
|
self._cache_clear()
|
|
return r.json()
|
|
|
|
async def set_issue_state(
|
|
self, owner: str, repo: str, issue_number: int, state: str
|
|
) -> dict:
|
|
"""Open or close an issue in Gitea (state: "open" | "closed")."""
|
|
if state not in ("open", "closed"):
|
|
raise ValueError(f"Invalid issue state: {state}")
|
|
r = await self.client.patch(
|
|
f"/repos/{owner}/{repo}/issues/{issue_number}", json={"state": state}
|
|
)
|
|
r.raise_for_status()
|
|
self._cache_clear()
|
|
return r.json()
|
|
|
|
async def create_branch(
|
|
self, owner: str, repo: str, new_branch: str, old_branch: str
|
|
) -> dict:
|
|
"""Create a branch from another one (issue #3)."""
|
|
return await self._request(
|
|
"POST",
|
|
f"/repos/{owner}/{repo}/branches",
|
|
json={
|
|
"new_branch_name": new_branch,
|
|
"old_branch_name": old_branch,
|
|
},
|
|
)
|
|
|
|
async def list_branches(self, owner: str, repo: str) -> list[dict]:
|
|
"""All branches of a repo (code links, issue #1)."""
|
|
return await self._get_all_pages(f"/repos/{owner}/{repo}/branches")
|
|
|
|
async def list_pulls(
|
|
self, owner: str, repo: str, state: str = "all"
|
|
) -> list[dict]:
|
|
"""All pull requests of a repo (code links, issue #1)."""
|
|
return await self._get_all_pages(
|
|
f"/repos/{owner}/{repo}/pulls", state=state
|
|
)
|
|
|
|
async def list_commits(
|
|
self, owner: str, repo: str, limit: int = 75
|
|
) -> list[dict]:
|
|
"""Most recent commits of a repo (code links, issue #1)."""
|
|
data = await self._get(f"/repos/{owner}/{repo}/commits", limit=limit)
|
|
return data if isinstance(data, list) else []
|
|
|
|
async def set_issue_assignees(
|
|
self, owner: str, repo: str, issue_number: int, logins: list[str]
|
|
) -> dict:
|
|
"""Replace the assignee set of an issue (triage "assign me")."""
|
|
r = await self.client.patch(
|
|
f"/repos/{owner}/{repo}/issues/{issue_number}",
|
|
json={"assignees": logins},
|
|
)
|
|
r.raise_for_status()
|
|
self._cache_clear()
|
|
return r.json()
|