49 lines
2.2 KiB
Python
49 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
|
|
class TodoistClient:
|
|
def __init__(self, api_key: str | None = None, base_url: str | None = None):
|
|
self.api_key = api_key or os.getenv("TODOIST_API_KEY")
|
|
self.base_url = (base_url or os.getenv("TODOIST_API_BASE_URL") or "https://api.todoist.com/rest/v2").rstrip("/")
|
|
self.project_id = os.getenv("TODOIST_PROJECT_ID")
|
|
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return bool(self.api_key)
|
|
|
|
def _headers(self) -> dict[str, str]:
|
|
if not self.api_key:
|
|
raise RuntimeError("TODOIST_API_KEY is not configured")
|
|
return {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}
|
|
|
|
async def create_task(self, *, content: str, description: str, due_string: str | None = None) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {"content": content, "description": description}
|
|
if self.project_id:
|
|
payload["project_id"] = self.project_id
|
|
if due_string:
|
|
payload["due_string"] = due_string
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
response = await client.post(f"{self.base_url}/tasks", headers=self._headers(), json=payload)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def update_task(self, task_id: str, *, content: str, description: str, due_string: str | None = None) -> None:
|
|
payload: dict[str, Any] = {"content": content, "description": description}
|
|
if due_string:
|
|
payload["due_string"] = due_string
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
response = await client.post(f"{self.base_url}/tasks/{task_id}", headers=self._headers(), json=payload)
|
|
response.raise_for_status()
|
|
|
|
async def add_comment(self, task_id: str, content: str) -> dict[str, Any]:
|
|
payload = {"task_id": task_id, "content": content}
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
response = await client.post(f"{self.base_url}/comments", headers=self._headers(), json=payload)
|
|
response.raise_for_status()
|
|
return response.json()
|