mobiledev/taskflow
TaskFlow, a team task board you run from the web or from Claude and ChatGPT. FastAPI + DataK3 + appid + Models + MCP.
A remix template · 0 remixes · Remix mobiledev/taskflow
app/taskflow/doors/web.py
"""The web door: JSON routes plus the board page. Every route calls a use case and nothing else."""
from pathlib import Path
from typing import Optional
from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from pydantic import BaseModel
from taskflow.application.use_cases import NotFound, TaskFlow
from taskflow.doors.identity import person_from
from taskflow.domain.model import Forbidden, Invalid, Person
PAGE = (Path(__file__).parent / "page.html").read_text(encoding="utf-8")
class NewBoard(BaseModel):
title: str
lists: Optional[list[str]] = None
class NewList(BaseModel):
name: str
before: Optional[str] = None
class RenameList(BaseModel):
new_name: str
class ConfirmTitle(BaseModel):
title: str
class NewMember(BaseModel):
email: str
role: str = "member"
class NewCard(BaseModel):
title: str
list_name: Optional[str] = None
description: str = ""
assignee: Optional[str] = None
due_date: Optional[str] = None
labels: Optional[list[str]] = None
class CardChanges(BaseModel):
"""Only the fields sent change; send null to clear assignee or due_date."""
title: Optional[str] = None
description: Optional[str] = None
assignee: Optional[str] = None
due_date: Optional[str] = None
labels: Optional[list[str]] = None
class Move(BaseModel):
list_name: str
index: Optional[int] = None
class Blocker(BaseModel):
blocked_by: str
class Items(BaseModel):
items: list[str]
class Tick(BaseModel):
done: bool = True
def build_router(app: TaskFlow) -> APIRouter:
r = APIRouter()
def me(request: Request) -> Person:
p = person_from(request.headers)
if p is None:
raise HTTPException(401, "sign in first: /.dodil/auth/login")
return p
def run(request: Request, fn, *args, **kwargs):
try:
return fn(me(request), *args, **kwargs)
except NotFound as e:
raise HTTPException(404, str(e))
except Forbidden as e:
raise HTTPException(403, str(e))
except Invalid as e:
raise HTTPException(422, str(e))
# --- boards ----------------------------------------------------------------------------
@r.get("/api/boards")
def my_boards(request: Request):
return run(request, app.my_boards)
@r.post("/api/boards", status_code=201)
def create_board(body: NewBoard, request: Request):
return run(request, app.create_board, body.title, body.lists)
@r.get("/api/boards/{board}")
def show_board(board: str, request: Request):
return run(request, app.show_board, board)
@r.post("/api/boards/{board}/lists", status_code=201)
def add_list(board: str, body: NewList, request: Request):
return run(request, app.add_list, board, body.name, body.before)
@r.patch("/api/boards/{board}/lists/{name}")
def rename_list(board: str, name: str, body: RenameList, request: Request):
return run(request, app.rename_list, board, name, body.new_name)
@r.delete("/api/boards/{board}/lists/{name}")
def delete_list(board: str, name: str, request: Request):
return run(request, app.delete_list, board, name)
@r.post("/api/boards/{board}/delete")
def delete_board(board: str, body: ConfirmTitle, request: Request):
return run(request, app.delete_board, board, body.title)
@r.get("/api/boards/{board}/members")
def members(board: str, request: Request):
return run(request, app.members, board)
@r.post("/api/boards/{board}/members", status_code=201)
def add_member(board: str, body: NewMember, request: Request):
return run(request, app.add_member, board, body.email, body.role)
@r.delete("/api/boards/{board}/members/{email}")
def remove_member(board: str, email: str, request: Request):
return run(request, app.remove_member, board, email)
@r.post("/api/boards/{board}/cards", status_code=201)
def add_card(board: str, body: NewCard, request: Request):
return run(request, app.add_card, board, body.title, body.list_name, body.description,
body.assignee, body.due_date, body.labels)
@r.get("/api/boards/{board}/activity")
def board_activity(board: str, request: Request, limit: int = 30):
return run(request, app.board_activity, board, limit)
@r.get("/api/boards/{board}/summary")
def board_summary(board: str, request: Request):
return run(request, app.board_summary, board)
# --- cards -----------------------------------------------------------------------------
@r.get("/api/cards/{card}")
def show_card(card: str, request: Request):
return run(request, app.show_card, card)
@r.patch("/api/cards/{card}")
def update_card(card: str, body: CardChanges, request: Request):
return run(request, app.update_card, card, None, **body.model_dump(exclude_unset=True))
@r.delete("/api/cards/{card}")
def delete_card(card: str, request: Request):
return run(request, app.delete_card, card)
@r.post("/api/cards/{card}/move")
def move_card(card: str, body: Move, request: Request):
return run(request, app.move_card, card, body.list_name, None, body.index)
@r.get("/api/cards/{card}/history")
def card_history(card: str, request: Request):
return run(request, app.card_history, card)
@r.post("/api/cards/{card}/blockers")
def block_card(card: str, body: Blocker, request: Request):
return run(request, app.block_card, card, body.blocked_by)
@r.delete("/api/cards/{card}/blockers/{blocker}")
def unblock_card(card: str, blocker: str, request: Request):
return run(request, app.unblock_card, card, blocker)
@r.get("/api/cards/{card}/blocking")
def what_blocks(card: str, request: Request):
return run(request, app.what_blocks, card)
@r.post("/api/cards/{card}/checklist")
def add_checklist(card: str, body: Items, request: Request):
return run(request, app.add_checklist, card, body.items)
@r.post("/api/cards/{card}/checklist/ai")
def checklist_from_ai(card: str, request: Request):
return run(request, app.checklist_from_ai, card)
@r.post("/api/cards/{card}/checklist/{item}")
def tick(card: str, item: int, body: Tick, request: Request):
return run(request, app.tick, card, item, body.done)
@r.get("/api/cards/{card}/similar")
def similar_cards(card: str, request: Request):
return run(request, app.similar_cards, card)
# --- across boards ---------------------------------------------------------------------
@r.get("/api/me/cards")
def my_cards(request: Request, due_within_days: Optional[int] = None):
return run(request, app.my_cards, due_within_days)
@r.get("/api/overdue")
def overdue(request: Request, board: Optional[str] = None):
return run(request, app.overdue, board)
@r.get("/api/me")
def whoami(request: Request):
return {"email": me(request).email}
@r.get("/", response_class=HTMLResponse)
def page(request: Request):
if person_from(request.headers) is None:
return RedirectResponse("/.dodil/auth/login?return_to=/", status_code=303)
return PAGE
return r