Spark
71881514c9
Gitea 1.27.3 renvoie "pull_request": null sur les issues ordinaires: _is_issue testait la présence de la clé et classait chaque vraie issue comme un pull request. Board, Focus et briefing affichaient zéro carte contre un vrai Gitea (le dashboard, basé sur les compteurs de dépôts, semblait normal, d'où le signalement « la nouvelle issue n'apparaît pas »). Le filtre teste maintenant la valeur. Vérifié en réel: board bruno/flowdeck 0 -> 6 cartes.
343 lines
12 KiB
Python
343 lines
12 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 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; 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}"
|
|
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()
|