fix(backend): consolidation 0.2.0 — config, DB_PATH, async, pagination, auth, synchro bidirectionnelle
- config.py: réglages centralisés par variables d'environnement - models: KANBANTE_DB_PATH enfin honorée (C1), migrations idempotentes (position, notes.updated_at, nettoyage colonnes inconnues — I4), ordre des cartes persisté (M5), réconciliation avec Gitea - gitea_client: httpx asynchrone, pagination complète (I2, plafond serveur de 50), uid résolu via /user, en-tête Authorization seulement si jeton, cache court, create_issue / set_issue_state, .env racine - main: lifespan au lieu de on_event (M4), Basic Auth optionnelle + CSRF double-submit (C2), write-back done<->fermé (C3), classification par fenêtre glissante + drapeau archived (I1), erreurs Gitea en 502 propre (I5), /api/readiness réel, PR exclues du board (M6), codes HTTP explicites et limite de taille des notes (M7)
This commit is contained in:
Spark
+187
-40
@@ -1,62 +1,187 @@
|
||||
"""
|
||||
Gitea API client for Kanbante.
|
||||
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 os
|
||||
import logging
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv(os.path.expanduser("~/workspace/.env"))
|
||||
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 Gitea API client for Kanbante."""
|
||||
"""Lightweight async Gitea API client for Kanbante."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.base_url = os.getenv("GITEA_URL", "https://git.dracodev.net")
|
||||
self.token = os.getenv("GITEA_TOKEN", "")
|
||||
self.client = httpx.Client(
|
||||
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={
|
||||
"Authorization": f"token {self.token}",
|
||||
"Accept": "application/json",
|
||||
},
|
||||
headers=headers,
|
||||
timeout=15.0,
|
||||
transport=transport,
|
||||
)
|
||||
self._user: dict | None = None
|
||||
self._cache: dict[str, tuple[float, Any]] = {}
|
||||
|
||||
def _get(self, path: str, **params: Any) -> dict | list:
|
||||
r = self.client.get(path, params=params)
|
||||
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()
|
||||
|
||||
def get_user_repos(self, username: str = "bruno") -> list[dict]:
|
||||
"""Get all repos for a user."""
|
||||
return self._get(f"/users/{username}/repos", limit=100)
|
||||
async def _get_all_pages(self, path: str, **params: Any) -> list[dict]:
|
||||
"""Fetch every page of a list endpoint.
|
||||
|
||||
def search_repos(self, uid: int = 1) -> list[dict]:
|
||||
"""Search all repos visible to user."""
|
||||
result = self._get("/repos/search", uid=uid, limit=200)
|
||||
if isinstance(result, dict) and "data" in result:
|
||||
return result["data"]
|
||||
return result if isinstance(result, list) else []
|
||||
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)
|
||||
|
||||
def get_org_repos(self, org: str) -> list[dict]:
|
||||
"""Get repos for an organization."""
|
||||
return self._get(f"/orgs/{org}/repos", limit=50)
|
||||
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 get_repo_issues(
|
||||
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()
|
||||
|
||||
# ── 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."""
|
||||
return self._get(
|
||||
f"/repos/{owner}/{repo}/issues",
|
||||
state=state,
|
||||
limit=100,
|
||||
)
|
||||
"""Get issues for a repo (paginated, cached).
|
||||
|
||||
def get_all_issues(
|
||||
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."""
|
||||
@@ -65,12 +190,34 @@ class GiteaClient:
|
||||
full_name = repo["full_name"]
|
||||
owner, name = full_name.split("/", 1)
|
||||
try:
|
||||
issues = self.get_repo_issues(owner, name, state=state)
|
||||
issues_by_repo[full_name] = issues
|
||||
except Exception:
|
||||
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
|
||||
|
||||
def get_user(self) -> dict:
|
||||
"""Get authenticated user info."""
|
||||
return self._get("/user")
|
||||
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()
|
||||
|
||||
Reference in New Issue
Block a user