Initial WhatToDo app

Self-hosted personal task list: FastAPI + SQLite in Docker, mobile-friendly PWA frontend, deadlines, priorities, categories, tags, notes, checklists, recurring tasks, single-password login, and JSON export/import. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-07 11:43:56 +02:00
commit c5b049edba
23 changed files with 2045 additions and 0 deletions
@@ -0,0 +1,527 @@
+import calendar
+import hashlib
+import hmac
+import json
+import os
+import sqlite3
+import time
+from collections import defaultdict, deque
+from contextlib import asynccontextmanager
+from datetime import date, datetime, timedelta, timezone
+from pathlib import Path
+from typing import Literal, Optional
+
+from fastapi import Depends, FastAPI, HTTPException, Request, Response
+from fastapi.responses import JSONResponse
+from fastapi.staticfiles import StaticFiles
+from pydantic import BaseModel, Field, field_validator
+
+from . import db
+
+# --- Configuration -----------------------------------------------------------
+
+APP_PASSWORD = os.environ.get("APP_PASSWORD", "")
+SECRET_KEY = os.environ.get("SECRET_KEY", "")
+COOKIE_SECURE = os.environ.get("COOKIE_SECURE", "false").lower() in ("1", "true", "yes")
+SESSION_DAYS = int(os.environ.get("SESSION_DAYS", "30"))
+
+if not APP_PASSWORD or not SECRET_KEY or "change-me" in (APP_PASSWORD, SECRET_KEY):
+    raise SystemExit("APP_PASSWORD and SECRET_KEY must be set to real values (see .env.example)")
+
+COOKIE_NAME = "whattodo_session"
+CSRF_HEADER = "x-whattodo"
+EXPORT_VERSION = 1
+STATIC_DIR = Path(__file__).parent / "static"
+
+# Changing the password (or secret) invalidates all existing sessions.
+_SESSION_KEY = hashlib.sha256(f"{SECRET_KEY}\0{APP_PASSWORD}".encode()).digest()
+
+
+@asynccontextmanager
+async def lifespan(_app: FastAPI):
+    db.init()
+    yield
+
+
+app = FastAPI(title="WhatToDo", lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None)
+
+
+# --- Sessions & security ------------------------------------------------------
+
+def make_token() -> str:
+    ts = str(int(time.time()))
+    sig = hmac.new(_SESSION_KEY, ts.encode(), hashlib.sha256).hexdigest()
+    return f"{ts}.{sig}"
+
+
+def valid_token(token: Optional[str]) -> bool:
+    if not token or "." not in token:
+        return False
+    ts, sig = token.split(".", 1)
+    expected = hmac.new(_SESSION_KEY, ts.encode(), hashlib.sha256).hexdigest()
+    if not hmac.compare_digest(sig, expected):
+        return False
+    try:
+        return time.time() - int(ts) < SESSION_DAYS * 86400
+    except ValueError:
+        return False
+
+
+PUBLIC_API = {"/api/login", "/api/health"}
+CSP = (
+    "default-src 'self'; script-src 'self'; style-src 'self'; img-src 'self' data:; "
+    "connect-src 'self'; manifest-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'self'"
+)
+
+
+@app.middleware("http")
+async def guard(request: Request, call_next):
+    path = request.url.path
+    if path.startswith("/api/"):
+        if path not in PUBLIC_API and not valid_token(request.cookies.get(COOKIE_NAME)):
+            return JSONResponse({"detail": "Not logged in"}, status_code=401)
+        # Custom header on every write request: browsers can't send it cross-site without CORS.
+        if request.method not in ("GET", "HEAD", "OPTIONS") and request.headers.get(CSRF_HEADER) != "1":
+            return JSONResponse({"detail": "Missing X-WhatToDo header"}, status_code=403)
+    response = await call_next(request)
+    response.headers["X-Content-Type-Options"] = "nosniff"
+    response.headers["Referrer-Policy"] = "same-origin"
+    response.headers["X-Frame-Options"] = "DENY"
+    response.headers["Content-Security-Policy"] = CSP
+    if path.startswith("/api/"):
+        response.headers["Cache-Control"] = "no-store"
+    else:
+        response.headers.setdefault("Cache-Control", "no-cache")
+    return response
+
+
+_failed_logins: dict[str, deque] = defaultdict(deque)
+MAX_FAILS, FAIL_WINDOW = 10, 15 * 60
+
+
+class LoginIn(BaseModel):
+    password: str
+
+
+@app.post("/api/login")
+def login(body: LoginIn, request: Request, response: Response):
+    ip = request.client.host if request.client else "unknown"
+    fails, now = _failed_logins[ip], time.time()
+    while fails and now - fails[0] > FAIL_WINDOW:
+        fails.popleft()
+    if len(fails) >= MAX_FAILS:
+        raise HTTPException(429, "Too many failed attempts, try again in a few minutes")
+    if not hmac.compare_digest(body.password.encode(), APP_PASSWORD.encode()):
+        fails.append(now)
+        time.sleep(1)
+        raise HTTPException(401, "Wrong password")
+    fails.clear()
+    response.set_cookie(
+        COOKIE_NAME, make_token(), max_age=SESSION_DAYS * 86400,
+        httponly=True, samesite="lax", secure=COOKIE_SECURE, path="/",
+    )
+    return {"ok": True}
+
+
+@app.post("/api/logout")
+def logout(response: Response):
+    response.delete_cookie(COOKIE_NAME, path="/")
+    return {"ok": True}
+
+
+@app.get("/api/health")
+def health():
+    return {"ok": True}
+
+
+@app.get("/api/me")
+def me():
+    return {"ok": True}
+
+
+# --- Helpers -----------------------------------------------------------------
+
+def get_conn():
+    conn = db.connect()
+    try:
+        yield conn
+    finally:
+        conn.close()
+
+
+def now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat(timespec="seconds")
+
+
+def add_months(d: date, n: int) -> date:
+    m = d.month - 1 + n
+    y, m = d.year + m // 12, m % 12 + 1
+    return date(y, m, min(d.day, calendar.monthrange(y, m)[1]))
+
+
+def step(d: date, recurrence: str, n: int) -> date:
+    if recurrence == "daily":
+        return d + timedelta(days=n)
+    if recurrence == "weekly":
+        return d + timedelta(weeks=n)
+    if recurrence == "monthly":
+        return add_months(d, n)
+    return add_months(d, 12 * n)  # yearly
+
+
+def next_due(due: Optional[date], recurrence: str, n: int, today: date) -> date:
+    """Next occurrence after `today`, so a late completion doesn't create an already-overdue task."""
+    nxt = step(due or today, recurrence, n)
+    while nxt <= today:
+        nxt = step(nxt, recurrence, n)
+    return nxt
+
+
+def norm_tags(tags: list[str]) -> list[str]:
+    out: list[str] = []
+    for t in tags:
+        t = t.strip().lstrip("#").strip().lower()
+        if t and len(t) <= 40 and t not in out:
+            out.append(t)
+    return out[:20]
+
+
+def clean_title(v: Optional[str]) -> Optional[str]:
+    if v is None:
+        return v
+    v = v.strip()
+    if not v:
+        raise ValueError("must not be empty")
+    return v
+
+
+# --- Models ------------------------------------------------------------------
+
+Recurrence = Literal["none", "daily", "weekly", "monthly", "yearly"]
+HexColor = Field(pattern=r"^#[0-9a-fA-F]{6}$")
+
+
+class SubtaskIn(BaseModel):
+    title: str = Field(min_length=1, max_length=500)
+    done: bool = False
+
+
+class TaskIn(BaseModel):
+    title: str = Field(max_length=500)
+    notes: str = Field("", max_length=20000)
+    category_id: Optional[int] = None
+    priority: int = Field(0, ge=0, le=3)
+    due_date: Optional[date] = None
+    tags: list[str] = []
+    recurrence: Recurrence = "none"
+    recur_interval: int = Field(1, ge=1, le=365)
+    subtasks: list[SubtaskIn] = Field([], max_length=200)
+
+    _title = field_validator("title")(clean_title)
+
+
+class TaskPatch(BaseModel):
+    title: Optional[str] = Field(None, max_length=500)
+    notes: Optional[str] = Field(None, max_length=20000)
+    category_id: Optional[int] = None
+    priority: Optional[int] = Field(None, ge=0, le=3)
+    due_date: Optional[date] = None
+    tags: Optional[list[str]] = None
+    recurrence: Optional[Recurrence] = None
+    recur_interval: Optional[int] = Field(None, ge=1, le=365)
+    subtasks: Optional[list[SubtaskIn]] = Field(None, max_length=200)
+
+    _title = field_validator("title")(clean_title)
+
+
+class CompleteIn(BaseModel):
+    today: Optional[date] = None  # the browser's local date
+
+
+class CategoryIn(BaseModel):
+    name: str = Field(min_length=1, max_length=50)
+    color: str = HexColor
+    position: Optional[int] = None
+
+
+class CategoryPatch(BaseModel):
+    name: Optional[str] = Field(None, min_length=1, max_length=50)
+    color: Optional[str] = Field(None, pattern=r"^#[0-9a-fA-F]{6}$")
+    position: Optional[int] = None
+
+
+class ExportCategory(BaseModel):
+    id: int
+    name: str = Field(min_length=1, max_length=50)
+    color: str = HexColor
+    position: int = 0
+
+
+class ExportTask(TaskIn):
+    done: bool = False
+    done_at: Optional[str] = None
+    created_at: Optional[str] = None
+    updated_at: Optional[str] = None
+
+
+class ExportData(BaseModel):
+    app: str
+    version: int
+    categories: list[ExportCategory] = []
+    tasks: list[ExportTask] = []
+
+
+# --- Data access -------------------------------------------------------------
+
+def row_to_task(r: sqlite3.Row, subtasks: list[dict]) -> dict:
+    return {
+        "id": r["id"],
+        "title": r["title"],
+        "notes": r["notes"],
+        "category_id": r["category_id"],
+        "priority": r["priority"],
+        "due_date": r["due_date"],
+        "tags": json.loads(r["tags"]),
+        "recurrence": r["recurrence"],
+        "recur_interval": r["recur_interval"],
+        "done": bool(r["done"]),
+        "done_at": r["done_at"],
+        "created_at": r["created_at"],
+        "updated_at": r["updated_at"],
+        "subtasks": subtasks,
+    }
+
+
+def load_tasks(conn: sqlite3.Connection, task_id: Optional[int] = None) -> list[dict]:
+    where, params = ("WHERE id = ?", (task_id,)) if task_id is not None else ("", ())
+    sub_where = "WHERE task_id = ?" if task_id is not None else ""
+    subs: dict[int, list[dict]] = defaultdict(list)
+    for s in conn.execute(f"SELECT * FROM subtasks {sub_where} ORDER BY position, id", params):
+        subs[s["task_id"]].append({"id": s["id"], "title": s["title"], "done": bool(s["done"])})
+    rows = conn.execute(f"SELECT * FROM tasks {where} ORDER BY id", params).fetchall()
+    return [row_to_task(r, subs[r["id"]]) for r in rows]
+
+
+def get_task(conn: sqlite3.Connection, task_id: int) -> dict:
+    tasks = load_tasks(conn, task_id)
+    if not tasks:
+        raise HTTPException(404, "Task not found")
+    return tasks[0]
+
+
+def ensure_category(conn: sqlite3.Connection, category_id: Optional[int]) -> None:
+    if category_id is not None and not conn.execute(
+        "SELECT 1 FROM categories WHERE id = ?", (category_id,)
+    ).fetchone():
+        raise HTTPException(400, "Unknown category")
+
+
+def write_subtasks(conn: sqlite3.Connection, task_id: int, subtasks: list[dict]) -> None:
+    conn.execute("DELETE FROM subtasks WHERE task_id = ?", (task_id,))
+    for i, s in enumerate(subtasks):
+        conn.execute(
+            "INSERT INTO subtasks (task_id, title, done, position) VALUES (?, ?, ?, ?)",
+            (task_id, s["title"].strip(), int(s["done"]), i),
+        )
+
+
+def insert_task(conn: sqlite3.Connection, t: dict, done: bool = False, done_at: Optional[str] = None,
+                created_at: Optional[str] = None, updated_at: Optional[str] = None) -> int:
+    now = now_iso()
+    due = t.get("due_date")
+    cur = conn.execute(
+        """INSERT INTO tasks (title, notes, category_id, priority, due_date, tags, recurrence,
+                              recur_interval, done, done_at, created_at, updated_at)
+           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+        (
+            t["title"], t.get("notes", ""), t.get("category_id"), t.get("priority", 0),
+            due.isoformat() if isinstance(due, date) else due,
+            json.dumps(norm_tags(t.get("tags", []))), t.get("recurrence", "none"),
+            t.get("recur_interval", 1), int(done), done_at, created_at or now, updated_at or now,
+        ),
+    )
+    write_subtasks(conn, cur.lastrowid, t.get("subtasks", []))
+    return cur.lastrowid
+
+
+# --- Category endpoints ------------------------------------------------------
+
+def list_categories(conn: sqlite3.Connection) -> list[dict]:
+    return [dict(r) for r in conn.execute("SELECT * FROM categories ORDER BY position, id")]
+
+
+@app.get("/api/categories")
+def get_categories(conn=Depends(get_conn)):
+    return list_categories(conn)
+
+
+@app.post("/api/categories", status_code=201)
+def create_category(body: CategoryIn, conn=Depends(get_conn)):
+    pos = body.position
+    if pos is None:
+        pos = conn.execute("SELECT COALESCE(MAX(position), -1) + 1 FROM categories").fetchone()[0]
+    try:
+        with conn:
+            cur = conn.execute(
+                "INSERT INTO categories (name, color, position) VALUES (?, ?, ?)",
+                (body.name.strip(), body.color, pos),
+            )
+    except sqlite3.IntegrityError:
+        raise HTTPException(409, "A category with that name already exists")
+    return dict(conn.execute("SELECT * FROM categories WHERE id = ?", (cur.lastrowid,)).fetchone())
+
+
+@app.patch("/api/categories/{category_id}")
+def update_category(category_id: int, body: CategoryPatch, conn=Depends(get_conn)):
+    ensure_category(conn, category_id)
+    fields = body.model_dump(exclude_unset=True, exclude_none=True)
+    if "name" in fields:
+        fields["name"] = fields["name"].strip()
+    if fields:
+        try:
+            with conn:
+                conn.execute(
+                    f"UPDATE categories SET {', '.join(f'{k} = ?' for k in fields)} WHERE id = ?",
+                    (*fields.values(), category_id),
+                )
+        except sqlite3.IntegrityError:
+            raise HTTPException(409, "A category with that name already exists")
+    return dict(conn.execute("SELECT * FROM categories WHERE id = ?", (category_id,)).fetchone())
+
+
+@app.delete("/api/categories/{category_id}", status_code=204)
+def delete_category(category_id: int, conn=Depends(get_conn)):
+    with conn:
+        conn.execute("DELETE FROM categories WHERE id = ?", (category_id,))
+    return Response(status_code=204)
+
+
+# --- Task endpoints ----------------------------------------------------------
+
+@app.get("/api/tasks")
+def get_tasks(conn=Depends(get_conn)):
+    return load_tasks(conn)
+
+
+@app.post("/api/tasks", status_code=201)
+def create_task(body: TaskIn, conn=Depends(get_conn)):
+    ensure_category(conn, body.category_id)
+    with conn:
+        task_id = insert_task(conn, body.model_dump())
+    return get_task(conn, task_id)
+
+
+@app.patch("/api/tasks/{task_id}")
+def update_task(task_id: int, body: TaskPatch, conn=Depends(get_conn)):
+    get_task(conn, task_id)
+    fields = body.model_dump(exclude_unset=True)
+    subtasks = fields.pop("subtasks", None)
+    for key in ("title", "notes", "priority", "tags", "recurrence", "recur_interval"):
+        if key in fields and fields[key] is None:
+            del fields[key]  # these columns can't be cleared, only changed
+    if "category_id" in fields:
+        ensure_category(conn, fields["category_id"])
+    if "tags" in fields:
+        fields["tags"] = json.dumps(norm_tags(fields["tags"]))
+    if fields.get("due_date") is not None:
+        fields["due_date"] = fields["due_date"].isoformat()
+    fields["updated_at"] = now_iso()
+    with conn:
+        conn.execute(
+            f"UPDATE tasks SET {', '.join(f'{k} = ?' for k in fields)} WHERE id = ?",
+            (*fields.values(), task_id),
+        )
+        if subtasks is not None:
+            write_subtasks(conn, task_id, subtasks)
+    return get_task(conn, task_id)
+
+
+@app.delete("/api/tasks/{task_id}", status_code=204)
+def delete_task(task_id: int, conn=Depends(get_conn)):
+    with conn:
+        conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
+    return Response(status_code=204)
+
+
+@app.post("/api/tasks/{task_id}/complete")
+def complete_task(task_id: int, body: CompleteIn, conn=Depends(get_conn)):
+    """Mark done. A recurring task also spawns its next occurrence (with subtasks unchecked)."""
+    task = get_task(conn, task_id)
+    if task["done"]:
+        return {"task": task, "next": None}
+    today = body.today or date.today()
+    next_id = None
+    with conn:
+        now = now_iso()
+        conn.execute("UPDATE tasks SET done = 1, done_at = ?, updated_at = ? WHERE id = ?", (now, now, task_id))
+        if task["recurrence"] != "none":
+            due = date.fromisoformat(task["due_date"]) if task["due_date"] else None
+            nxt = dict(task)
+            nxt["due_date"] = next_due(due, task["recurrence"], task["recur_interval"], today)
+            nxt["subtasks"] = [{"title": s["title"], "done": False} for s in task["subtasks"]]
+            next_id = insert_task(conn, nxt)
+    return {"task": get_task(conn, task_id), "next": get_task(conn, next_id) if next_id else None}
+
+
+@app.post("/api/tasks/{task_id}/reopen")
+def reopen_task(task_id: int, conn=Depends(get_conn)):
+    get_task(conn, task_id)
+    with conn:
+        conn.execute(
+            "UPDATE tasks SET done = 0, done_at = NULL, updated_at = ? WHERE id = ?", (now_iso(), task_id)
+        )
+    return get_task(conn, task_id)
+
+
+# --- Export / import ---------------------------------------------------------
+
+@app.get("/api/export")
+def export_data(conn=Depends(get_conn)):
+    stamp = datetime.now().strftime("%Y%m%d-%H%M")
+    data = {
+        "app": "whattodo",
+        "version": EXPORT_VERSION,
+        "exported_at": now_iso(),
+        "categories": list_categories(conn),
+        "tasks": load_tasks(conn),
+    }
+    return Response(
+        json.dumps(data, indent=2, ensure_ascii=False),
+        media_type="application/json",
+        headers={"Content-Disposition": f'attachment; filename="whattodo-export-{stamp}.json"'},
+    )
+
+
+@app.post("/api/import")
+def import_data(data: ExportData, mode: Literal["replace", "merge"] = "merge", conn=Depends(get_conn)):
+    """replace: wipe everything and load the file. merge: add the file's tasks, matching categories by name."""
+    if data.app != "whattodo" or data.version > EXPORT_VERSION:
+        raise HTTPException(400, "This file is not a WhatToDo export (or is from a newer version)")
+    with conn:
+        if mode == "replace":
+            conn.execute("DELETE FROM subtasks")
+            conn.execute("DELETE FROM tasks")
+            conn.execute("DELETE FROM categories")
+        existing = {r["name"].lower(): r["id"] for r in conn.execute("SELECT id, name FROM categories")}
+        next_pos = conn.execute("SELECT COALESCE(MAX(position), -1) + 1 FROM categories").fetchone()[0]
+        cat_map: dict[int, int] = {}
+        for c in sorted(data.categories, key=lambda c: c.position):
+            key = c.name.strip().lower()
+            if key not in existing:
+                cur = conn.execute(
+                    "INSERT INTO categories (name, color, position) VALUES (?, ?, ?)",
+                    (c.name.strip(), c.color, next_pos),
+                )
+                existing[key] = cur.lastrowid
+                next_pos += 1
+            cat_map[c.id] = existing[key]
+        for t in data.tasks:
+            fields = t.model_dump()
+            fields["category_id"] = cat_map.get(t.category_id) if t.category_id is not None else None
+            insert_task(conn, fields, done=t.done, done_at=t.done_at,
+                        created_at=t.created_at, updated_at=t.updated_at)
+    return {"ok": True, "mode": mode, "categories": len(data.categories), "tasks": len(data.tasks)}
+
+
+# Static frontend (must be mounted last so it doesn't shadow /api routes).
+app.mount("/", StaticFiles(directory=STATIC_DIR, html=True), name="static")