feat(integration): Gitea + Vikunja + CI 클라이언트 구현 #task-192 #task-193

integrations/vikunja_client.py

@@ -0,0 +1,146 @@
+"""Vikunja API 클라이언트.
+
+plan.variet.net API를 통해 태스크를 관리합니다.
+vikunja_helper.py의 async 버전 — safe_update 패턴 유지.
+"""
+
+import logging
+from typing import Optional
+
+import httpx
+
+import config
+
+logger = logging.getLogger("variet.vikunja")
+
+
+class VikunjaClient:
+    """Vikunja REST API 비동기 클라이언트."""
+
+    def __init__(
+        self,
+        base_url: str = None,
+        token: str = None,
+        project_id: int = None,
+    ):
+        self.base_url = (base_url or config.VIKUNJA_URL).rstrip("/")
+        self.api_url = f"{self.base_url}/api/v1"
+        self.token = token or config.VIKUNJA_TOKEN
+        self.project_id = project_id or config.VIKUNJA_PROJECT_ID
+        self.headers = {
+            "Authorization": f"Bearer {self.token}",
+            "Content-Type": "application/json",
+        }
+
+    async def _get(self, path: str, params: dict = None) -> dict | list:
+        async with httpx.AsyncClient(timeout=30) as client:
+            resp = await client.get(
+                f"{self.api_url}{path}",
+                headers=self.headers,
+                params=params,
+            )
+            resp.raise_for_status()
+            return resp.json()
+
+    async def _post(self, path: str, data: dict = None) -> dict:
+        async with httpx.AsyncClient(timeout=30) as client:
+            resp = await client.post(
+                f"{self.api_url}{path}",
+                headers=self.headers,
+                json=data,
+            )
+            resp.raise_for_status()
+            return resp.json()
+
+    async def _put(self, path: str, data: dict = None) -> dict:
+        async with httpx.AsyncClient(timeout=30) as client:
+            resp = await client.put(
+                f"{self.api_url}{path}",
+                headers=self.headers,
+                json=data,
+            )
+            resp.raise_for_status()
+            return resp.json()
+
+    # === Tasks ===
+
+    async def list_tasks(self, filter_: str = "all") -> list[dict]:
+        """프로젝트 태스크 목록 조회."""
+        tasks = await self._get(
+            f"/projects/{self.project_id}/tasks",
+            params={"per_page": 100},
+        )
+
+        if filter_ == "todo":
+            tasks = [t for t in tasks if not t["done"]]
+        elif filter_ == "done":
+            tasks = [t for t in tasks if t["done"]]
+
+        tasks.sort(key=lambda t: t["id"])
+        return [
+            {
+                "id": t["id"],
+                "title": t["title"],
+                "description": (t.get("description") or "")[:100],
+                "done": t["done"],
+                "labels": [l["title"] for l in (t.get("labels") or [])],
+            }
+            for t in tasks
+        ]
+
+    async def get_task(self, task_id: int) -> dict:
+        """태스크 상세 조회."""
+        return await self._get(f"/tasks/{task_id}")
+
+    async def safe_update_task(self, task_id: int, updates: dict) -> dict:
+        """안전한 태스크 업데이트 — GET → 보존 → POST.
+
+        Vikunja API는 POST 시 누락된 필드를 빈값으로 덮어쓰므로
+        반드시 기존 필드를 보존해야 합니다.
+        """
+        task = await self.get_task(task_id)
+        safe_body = {
+            "title": task.get("title", ""),
+            "description": task.get("description", ""),
+            "priority": task.get("priority", 0),
+            "done": task.get("done", False),
+        }
+        safe_body.update(updates)
+        result = await self._post(f"/tasks/{task_id}", safe_body)
+        logger.info(f"태스크 #{task_id} 업데이트: {updates}")
+        return result
+
+    async def mark_done(self, task_id: int) -> dict:
+        """태스크 완료 처리."""
+        result = await self.safe_update_task(task_id, {"done": True})
+        logger.info(f"태스크 #{task_id} 완료 처리")
+        return result
+
+    async def create_task(
+        self,
+        title: str,
+        description: str = "",
+        done: bool = False,
+    ) -> dict:
+        """새 태스크 생성."""
+        payload = {"title": title, "description": description}
+        result = await self._put(
+            f"/projects/{self.project_id}/tasks",
+            data=payload,
+        )
+        task_id = result["id"]
+        logger.info(f"태스크 #{task_id} 생성: {title}")
+
+        if done:
+            result = await self.safe_update_task(task_id, {"done": True})
+
+        return result
+
+    async def add_comment(self, task_id: int, comment: str) -> dict:
+        """태스크에 코멘트 추가."""
+        result = await self._put(
+            f"/tasks/{task_id}/comments",
+            data={"comment": comment},
+        )
+        logger.info(f"태스크 #{task_id} 코멘트 추가")
+        return result