147 lines
4.7 KiB
Python
147 lines
4.7 KiB
Python
"""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
|