Fix blumeops-tasks for Todoist API v1 migration (#155)

## Summary
- Migrate from deprecated Todoist REST API v2 (`410 Gone`) to new unified API v1
- Add cursor-based pagination for project and task listing endpoints
- Switch 1Password credential retrieval from `op item get --fields` to `op read`

## Testing
- [x] `mise run blumeops-tasks` returns all 9 tasks successfully
- [x] Pre-commit hooks pass

Reviewed-on: https://forge.ops.eblu.me/eblume/blumeops/pulls/155
This commit is contained in:
Erich Blume 2026-02-11 14:33:37 -08:00
commit faebc98c3c
2 changed files with 31 additions and 22 deletions

View file

@ -0,0 +1 @@
Fix blumeops-tasks: migrate from deprecated Todoist REST API v2 to API v1, handle cursor-based pagination, and use `op read` for 1Password credential retrieval.

View file

@ -28,7 +28,7 @@ import httpx
from rich.console import Console from rich.console import Console
from rich.text import Text from rich.text import Text
TODOIST_API_BASE = "https://api.todoist.com/rest/v2" TODOIST_API_BASE = "https://api.todoist.com/api/v1"
PROJECT_NAME = "Blumeops" PROJECT_NAME = "Blumeops"
# Priority mapping: Todoist API uses 1=normal(p4), 2=moderate(p3), 3=high(p2), 4=urgent(p1) # Priority mapping: Todoist API uses 1=normal(p4), 2=moderate(p3), 3=high(p2), 4=urgent(p1)
@ -40,17 +40,7 @@ PRIORITY_SORT_ORDER = {4: 1, 3: 2, 1: 3, 2: 4} # Lower = earlier
def get_todoist_token() -> str: def get_todoist_token() -> str:
"""Retrieve Todoist API token from 1Password.""" """Retrieve Todoist API token from 1Password."""
result = subprocess.run( result = subprocess.run(
[ ["op", "read", "op://vg6xf6vvfmoh5hqjjhlhbeoaie/c53h3xnmswhvexa5mntoyvhgpm/credential"],
"op",
"--vault",
"vg6xf6vvfmoh5hqjjhlhbeoaie",
"item",
"get",
"c53h3xnmswhvexa5mntoyvhgpm",
"--fields",
"credential",
"--reveal",
],
capture_output=True, capture_output=True,
text=True, text=True,
) )
@ -61,22 +51,40 @@ def get_todoist_token() -> str:
def get_project_id(client: httpx.Client, project_name: str) -> str: def get_project_id(client: httpx.Client, project_name: str) -> str:
"""Find project ID by name.""" """Find project ID by name."""
response = client.get(f"{TODOIST_API_BASE}/projects") cursor = None
while True:
params = {}
if cursor:
params["cursor"] = cursor
response = client.get(f"{TODOIST_API_BASE}/projects", params=params)
response.raise_for_status() response.raise_for_status()
projects = response.json() data = response.json()
for project in data.get("results", data if isinstance(data, list) else []):
for project in projects:
if project["name"] == project_name: if project["name"] == project_name:
return project["id"] return project["id"]
cursor = data.get("next_cursor") if isinstance(data, dict) else None
if not cursor:
break
raise RuntimeError(f"Project '{project_name}' not found in Todoist") raise RuntimeError(f"Project '{project_name}' not found in Todoist")
def get_tasks(client: httpx.Client, project_id: str) -> list[dict]: def get_tasks(client: httpx.Client, project_id: str) -> list[dict]:
"""Get all tasks for a project.""" """Get all tasks for a project."""
response = client.get(f"{TODOIST_API_BASE}/tasks", params={"project_id": project_id}) tasks = []
cursor = None
while True:
params = {"project_id": project_id}
if cursor:
params["cursor"] = cursor
response = client.get(f"{TODOIST_API_BASE}/tasks", params=params)
response.raise_for_status() response.raise_for_status()
return response.json() data = response.json()
tasks.extend(data.get("results", data if isinstance(data, list) else []))
cursor = data.get("next_cursor") if isinstance(data, dict) else None
if not cursor:
break
return tasks
def is_due(task: dict) -> bool: def is_due(task: dict) -> bool: