theaio/theaio-boards

Laneway, team task boards you can run from Claude or ChatGPT. FastAPI + DataK3 + appid + Models + MCP.

A remix template · 0 remixes · Remix theaio/theaio-boards

app/boards/doors/web.py

"""The web door: JSON routes plus one page. Every route calls a use case and nothing else.

Boards are addressed by id or title, cards by id or number (with ?board= when a number is on
several boards), exactly like the MCP tools.
"""
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 boards.application.use_cases import Boards, NotFound
from boards.doors.identity import person_from
from boards.domain.model import Forbidden, Invalid, Person

PAGE = (Path(__file__).parent / "page.html").read_text(encoding="utf-8")


class Title(BaseModel):
    title: str


class Name(BaseModel):
    name: str


class ListChange(BaseModel):
    name: Optional[str] = None
    position: Optional[int] = None


class CardIn(BaseModel):
    title: str
    list_name: str = ""
    description: str = ""


class CardChange(BaseModel):
    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
    position: Optional[int] = None


class Text(BaseModel):
    text: str


class Blocker(BaseModel):
    blocked_by: str


class Member(BaseModel):
    email: str
    role: str = "member"


class RoleIn(BaseModel):
    role: str


def build_router(boards: Boards) -> 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(fn, *args, **kwargs):
        try:
            return fn(*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, lists, people --------------------------------------------------------------
    @r.get("/api/boards")
    def my_boards(request: Request):
        return run(boards.my_boards, me(request))

    @r.post("/api/boards", status_code=201)
    def create_board(body: Title, request: Request):
        return run(boards.create_board, me(request), body.title)

    @r.get("/api/boards/{board}")
    def show_board(board: str, request: Request):
        return run(boards.show_board, me(request), board)

    @r.get("/api/boards/{board}/summary")
    def board_summary(board: str, request: Request):
        return run(boards.board_summary, me(request), board)

    @r.get("/api/boards/{board}/blocking")
    def what_is_blocking(board: str, request: Request):
        return run(boards.what_is_blocking, me(request), board)

    @r.post("/api/boards/{board}/lists", status_code=201)
    def add_list(board: str, body: Name, request: Request):
        return run(boards.add_list, me(request), board, body.name)

    @r.patch("/api/boards/{board}/lists/{list_ref}")
    def change_list(board: str, list_ref: str, body: ListChange, request: Request):
        person, out = me(request), None
        if body.name is not None:
            out = run(boards.rename_list, person, board, list_ref, body.name)
        if body.position is not None:
            out = run(boards.move_list, person, board, list_ref, body.position)
        if out is None:
            raise HTTPException(422, "send a new name or position")
        return out

    @r.post("/api/boards/{board}/members", status_code=201)
    def invite_member(board: str, body: Member, request: Request):
        return run(boards.invite_member, me(request), board, body.email, body.role)

    @r.patch("/api/boards/{board}/members/{email}")
    def set_member_role(board: str, email: str, body: RoleIn, request: Request):
        return run(boards.set_member_role, me(request), board, email, body.role)

    @r.delete("/api/boards/{board}/members/{email}")
    def remove_member(board: str, email: str, request: Request):
        return run(boards.remove_member, me(request), board, email)

    # --- cards ------------------------------------------------------------------------------
    @r.post("/api/boards/{board}/cards", status_code=201)
    def add_card(board: str, body: CardIn, request: Request):
        return run(boards.add_card, me(request), board, body.title, body.list_name,
                   body.description)

    @r.get("/api/cards/{card}")
    def show_card(card: str, request: Request, board: Optional[str] = None):
        return run(boards.show_card, me(request), card, board)

    @r.patch("/api/cards/{card}")
    def update_card(card: str, body: CardChange, request: Request, board: Optional[str] = None):
        return run(boards.update_card, me(request), card, board,
                   **body.model_dump(exclude_unset=True))

    @r.post("/api/cards/{card}/move")
    def move_card(card: str, body: Move, request: Request, board: Optional[str] = None):
        return run(boards.move_card, me(request), card, body.list_name, body.position, board)

    @r.post("/api/cards/{card}/comments", status_code=201)
    def comment_card(card: str, body: Text, request: Request, board: Optional[str] = None):
        return run(boards.comment_card, me(request), card, body.text, board)

    @r.get("/api/cards/{card}/history")
    def card_history(card: str, request: Request, board: Optional[str] = None):
        return run(boards.card_history, me(request), card, board)

    @r.post("/api/cards/{card}/blockers", status_code=201)
    def block_card(card: str, body: Blocker, request: Request, board: Optional[str] = None):
        return run(boards.block_card, me(request), card, body.blocked_by, board)

    @r.delete("/api/cards/{card}/blockers/{blocked_by}")
    def unblock_card(card: str, blocked_by: str, request: Request, board: Optional[str] = None):
        return run(boards.unblock_card, me(request), card, blocked_by, board)

    @r.get("/api/cards/{card}/similar")
    def similar_cards(card: str, request: Request, board: Optional[str] = None):
        return run(boards.similar_cards, me(request), card, board)

    @r.post("/api/cards/{card}/checklist")
    def suggest_checklist(card: str, request: Request, board: Optional[str] = None):
        return run(boards.suggest_checklist, me(request), card, board)

    # --- across boards ----------------------------------------------------------------------
    @r.get("/api/my-cards")
    def my_cards(request: Request, days: Optional[int] = None):
        return run(boards.my_cards, me(request), days)

    @r.get("/api/overdue")
    def list_overdue(request: Request):
        return run(boards.list_overdue, me(request))

    @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