bruno
336c7d8dfe
- Dashboard projets Gitea (user + orgs) - Board Kanban 5 colonnes avec drag & drop (SortableJS) - Sync bidirectionnelle colonnes ↔ labels Gitea - Filtres milestone/label/assignee - Notes Markdown par projet - API REST (/api/health, /api/move, /api/col-mapping…) - Colonnes custom, WIP limits - Thème clair/sombre - Cache Gitea TTL - Docker + docker-compose - Tests pytest - Docs: README, ARCHITECTURE, CHANGELOG, ROADMAP
209 lines
7.0 KiB
Python
209 lines
7.0 KiB
Python
"""FlowDeck — Gitea API client with caching."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class GiteaClient:
|
|
"""Async Gitea API v1 client with simple TTL cache."""
|
|
|
|
def __init__(self) -> None:
|
|
self._base = f"{settings.gitea_url}/api/v1"
|
|
self._headers = {"Authorization": f"token {settings.gitea_token}"}
|
|
self._cache: dict[str, tuple[datetime, Any]] = {}
|
|
self._ttl = timedelta(seconds=settings.gitea_cache_ttl)
|
|
|
|
# ── cache helper ──
|
|
|
|
def _cached(self, key: str) -> Any | None:
|
|
entry = self._cache.get(key)
|
|
if entry and entry[0] > datetime.now():
|
|
return entry[1]
|
|
return None
|
|
|
|
def _set_cache(self, key: str, value: Any) -> None:
|
|
self._cache[key] = (datetime.now() + self._ttl, value)
|
|
|
|
# ── repos ──
|
|
|
|
async def get_user_repos(self, page: int = 1, limit: int = 30) -> list[dict]:
|
|
"""List user repos (sorted by updated desc)."""
|
|
cache_key = f"repos:{page}:{limit}"
|
|
cached = self._cached(cache_key)
|
|
if cached:
|
|
return cached
|
|
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
resp = await client.get(
|
|
f"{self._base}/user/repos",
|
|
headers=self._headers,
|
|
params={"page": page, "limit": limit, "sort": "updated", "order": "desc"},
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
self._set_cache(cache_key, data)
|
|
return data
|
|
|
|
async def get_org_repos(self, org: str, page: int = 1, limit: int = 20) -> list[dict]:
|
|
"""List repos for an org."""
|
|
cache_key = f"org_repos:{org}:{page}:{limit}"
|
|
cached = self._cached(cache_key)
|
|
if cached:
|
|
return cached
|
|
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
resp = await client.get(
|
|
f"{self._base}/orgs/{org}/repos",
|
|
headers=self._headers,
|
|
params={"page": page, "limit": limit},
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
self._set_cache(cache_key, data)
|
|
return data
|
|
|
|
async def get_user_orgs(self) -> list[dict]:
|
|
"""List user orgs."""
|
|
cache_key = "user_orgs"
|
|
cached = self._cached(cache_key)
|
|
if cached:
|
|
return cached
|
|
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
resp = await client.get(
|
|
f"{self._base}/user/orgs", headers=self._headers
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
self._set_cache(cache_key, data)
|
|
return data
|
|
|
|
# ── issues ──
|
|
|
|
async def get_issues(
|
|
self, owner: str, repo: str, state: str = "all", page: int = 1, limit: int = 50
|
|
) -> list[dict]:
|
|
"""List issues for a repo."""
|
|
cache_key = f"issues:{owner}:{repo}:{state}:{page}:{limit}"
|
|
cached = self._cached(cache_key)
|
|
if cached:
|
|
return cached
|
|
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
resp = await client.get(
|
|
f"{self._base}/repos/{owner}/{repo}/issues",
|
|
headers=self._headers,
|
|
params={"state": state, "page": page, "limit": limit},
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
self._set_cache(cache_key, data)
|
|
return data
|
|
|
|
async def get_issue(self, owner: str, repo: str, issue_id: int) -> dict:
|
|
"""Get a single issue."""
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
resp = await client.get(
|
|
f"{self._base}/repos/{owner}/{repo}/issues/{issue_id}",
|
|
headers=self._headers,
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def update_issue_labels(
|
|
self, owner: str, repo: str, issue_id: int, labels: list[str]
|
|
) -> dict:
|
|
"""Replace issue labels."""
|
|
self._invalidate_issue_cache(owner, repo)
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
resp = await client.put(
|
|
f"{self._base}/repos/{owner}/{repo}/issues/{issue_id}",
|
|
headers=self._headers,
|
|
json={"labels": labels},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def close_issue(self, owner: str, repo: str, issue_id: int) -> dict:
|
|
"""Close an issue (move to Terminé)."""
|
|
self._invalidate_issue_cache(owner, repo)
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
resp = await client.patch(
|
|
f"{self._base}/repos/{owner}/{repo}/issues/{issue_id}",
|
|
headers=self._headers,
|
|
json={"state": "closed"},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def reopen_issue(self, owner: str, repo: str, issue_id: int) -> dict:
|
|
"""Reopen an issue."""
|
|
self._invalidate_issue_cache(owner, repo)
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
resp = await client.patch(
|
|
f"{self._base}/repos/{owner}/{repo}/issues/{issue_id}",
|
|
headers=self._headers,
|
|
json={"state": "open"},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
def _invalidate_issue_cache(self, owner: str, repo: str) -> None:
|
|
prefix = f"issues:{owner}:{repo}:"
|
|
to_delete = [k for k in self._cache if k.startswith(prefix)]
|
|
for k in to_delete:
|
|
del self._cache[k]
|
|
|
|
# ── labels ──
|
|
|
|
async def get_labels(self, owner: str, repo: str) -> list[dict]:
|
|
"""List repo labels."""
|
|
cache_key = f"labels:{owner}:{repo}"
|
|
cached = self._cached(cache_key)
|
|
if cached:
|
|
return cached
|
|
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
resp = await client.get(
|
|
f"{self._base}/repos/{owner}/{repo}/labels",
|
|
headers=self._headers,
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
self._set_cache(cache_key, data)
|
|
return data
|
|
|
|
# ── milestones ──
|
|
|
|
async def get_milestones(
|
|
self, owner: str, repo: str, state: str = "open"
|
|
) -> list[dict]:
|
|
"""List repo milestones."""
|
|
cache_key = f"milestones:{owner}:{repo}:{state}"
|
|
cached = self._cached(cache_key)
|
|
if cached:
|
|
return cached
|
|
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
resp = await client.get(
|
|
f"{self._base}/repos/{owner}/{repo}/milestones",
|
|
headers=self._headers,
|
|
params={"state": state},
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
self._set_cache(cache_key, data)
|
|
return data
|
|
|
|
# ── singleton ──
|
|
|
|
gitea = GiteaClient()
|