""" 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 logging import time 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]] = {} 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_get(self, key: str) -> Any | None: if self.cache_ttl <= 0: return None hit = self._cache.get(key) if hit and (time.monotonic() - hit[0]) < self.cache_ttl: return hit[1] return None 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() 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, cached).""" cached = self._cache_get("search_repos") if cached is not None: return cached 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) repos = await self._get_all_pages("/repos/search", **params) self._cache_put("search_repos", repos) return repos 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; each PR carries a "pull_request" key and callers filter on it. """ key = f"issues:{owner}/{repo}:{state}" cached = self._cache_get(key) if cached is not None: return cached issues = await self._get_all_pages( f"/repos/{owner}/{repo}/issues", state=state ) self._cache_put(key, issues) return issues 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}" cached = self._cache_get(key) if cached is not None: return cached milestones = await self._get_all_pages( f"/repos/{owner}/{repo}/milestones", state=state ) self._cache_put(key, milestones) return milestones 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()