Replace internal todos with Vikunja API wrapper
- New app/vikunja.py: get_tasks, create_task, update_task, delete_task - /todos page now reads from and writes to Vikunja (create, mark done, delete) - Auto-todos (clean build plate, filament low) post to Vikunja after DB commit - /api/todos router proxies to Vikunja - .env.install: VIKUNJA_URL, VIKUNJA_API_TOKEN, VIKUNJA_PROJECT_ID Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
81
app/vikunja.py
Normal file
81
app/vikunja.py
Normal file
@@ -0,0 +1,81 @@
|
||||
"""Vikunja task API client."""
|
||||
import logging
|
||||
import os
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger("pops")
|
||||
|
||||
|
||||
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_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:
|
||||
try:
|
||||
r = httpx.put(
|
||||
f"{_base()}/api/v1/projects/{_project_id()}/tasks",
|
||||
headers=_headers(),
|
||||
json={"title": title},
|
||||
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
|
||||
Reference in New Issue
Block a user