5dfa700916
- On create_task(), look up the 'POPS' label (create it if missing) and attach it to every task. Label ID is cached for the process lifetime. - Corrected VIKUNJA_URL typo (tasks → task) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
115 lines
3.0 KiB
Python
115 lines
3.0 KiB
Python
"""Vikunja task API client."""
|
|
import logging
|
|
import os
|
|
|
|
import httpx
|
|
|
|
logger = logging.getLogger("pops")
|
|
|
|
_pops_label_id: int | None = None
|
|
|
|
|
|
def _base() -> str:
|
|
return os.environ.get("VIKUNJA_URL", "").rstrip("/")
|
|
|
|
|
|
def _headers() -> dict:
|
|
return {
|
|
"Authorization": f"Bearer {os.environ.get('VIKUNJA_API_TOKEN', '')}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
|
|
def _project_id() -> int:
|
|
return int(os.environ.get("VIKUNJA_PROJECT_ID", 0))
|
|
|
|
|
|
def _get_or_create_pops_label() -> int | None:
|
|
"""Return the ID of the 'POPS' label, creating it if it doesn't exist."""
|
|
global _pops_label_id
|
|
if _pops_label_id is not None:
|
|
return _pops_label_id
|
|
try:
|
|
r = httpx.get(f"{_base()}/api/v1/labels", headers=_headers(), timeout=5)
|
|
r.raise_for_status()
|
|
for label in r.json() or []:
|
|
if label.get("title") == "POPS":
|
|
_pops_label_id = label["id"]
|
|
return _pops_label_id
|
|
# Not found — create it
|
|
r = httpx.put(
|
|
f"{_base()}/api/v1/labels",
|
|
headers=_headers(),
|
|
json={"title": "POPS"},
|
|
timeout=5,
|
|
)
|
|
r.raise_for_status()
|
|
_pops_label_id = r.json()["id"]
|
|
return _pops_label_id
|
|
except Exception:
|
|
logger.warning("Vikunja: failed to get or create POPS label")
|
|
return None
|
|
|
|
|
|
def get_tasks() -> list[dict]:
|
|
try:
|
|
r = httpx.get(
|
|
f"{_base()}/api/v1/projects/{_project_id()}/tasks",
|
|
headers=_headers(),
|
|
params={"sort_by": "created", "order_by": "desc"},
|
|
timeout=5,
|
|
)
|
|
r.raise_for_status()
|
|
return r.json() or []
|
|
except Exception:
|
|
logger.warning("Vikunja: failed to fetch tasks")
|
|
return []
|
|
|
|
|
|
def create_task(title: str) -> dict | None:
|
|
label_id = _get_or_create_pops_label()
|
|
payload: dict = {"title": title}
|
|
if label_id:
|
|
payload["labels"] = [{"id": label_id}]
|
|
try:
|
|
r = httpx.put(
|
|
f"{_base()}/api/v1/projects/{_project_id()}/tasks",
|
|
headers=_headers(),
|
|
json=payload,
|
|
timeout=5,
|
|
)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
except Exception:
|
|
logger.warning("Vikunja: failed to create task %r", title)
|
|
return None
|
|
|
|
|
|
def update_task(task_id: int, **fields) -> bool:
|
|
try:
|
|
r = httpx.post(
|
|
f"{_base()}/api/v1/tasks/{task_id}",
|
|
headers=_headers(),
|
|
json=fields,
|
|
timeout=5,
|
|
)
|
|
r.raise_for_status()
|
|
return True
|
|
except Exception:
|
|
logger.warning("Vikunja: failed to update task %s", task_id)
|
|
return False
|
|
|
|
|
|
def delete_task(task_id: int) -> bool:
|
|
try:
|
|
r = httpx.delete(
|
|
f"{_base()}/api/v1/tasks/{task_id}",
|
|
headers=_headers(),
|
|
timeout=5,
|
|
)
|
|
r.raise_for_status()
|
|
return True
|
|
except Exception:
|
|
logger.warning("Vikunja: failed to delete task %s", task_id)
|
|
return False
|