eduard/cobra-boards
Cobra Boards, team task boards you can run from Claude or ChatGPT. FastAPI + DataK3 + appid + Dodil Models + MCP.
A remix template · 1 remixes · Remix eduard/cobra-boards
app/boards/doors/web.py
"""The web door: a JSON API plus the board page. Every route calls a use case and nothing else,
so the page and Claude obey exactly the same rules."""
from pathlib import Path
from typing import List, Optional
from urllib.parse import quote
from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse, Response
from pydantic import BaseModel, ConfigDict, Field
from boards.application.use_cases import Boards, NotFound
from boards.domain.model import MAX_ATTACHMENT, Conflict, Forbidden, Invalid, Person
from boards.doors.identity import person_from
STATIC = Path(__file__).parent / "static"
class Title(BaseModel):
title: str
class NewBoard(BaseModel):
title: str
template: Optional[str] = None
class CommentIn(BaseModel):
body: str
class ReadIn(BaseModel):
ids: Optional[List[str]] = None
class ListIn(BaseModel):
name: str
is_done: bool = False
class ListPatch(BaseModel):
name: Optional[str] = None
is_done: Optional[bool] = None
position: Optional[float] = None
class CardIn(BaseModel):
model_config = ConfigDict(populate_by_name=True)
title: str
list_name: str = Field("", alias="list")
description: str = ""
assignee: Optional[str] = None
due_date: Optional[str] = None
labels: List[str] = []
class CardPatch(BaseModel):
expected_version: Optional[int] = None
title: Optional[str] = None
description: Optional[str] = None
assignee: Optional[str] = "__unset__"
due_date: Optional[str] = "__unset__"
labels: Optional[List[str]] = None
class Move(BaseModel):
model_config = ConfigDict(populate_by_name=True)
list_name: str = Field(alias="list")
before_card: Optional[str] = None
expected_version: Optional[int] = None
class Member(BaseModel):
email: str
role: str = "member"
class Blocker(BaseModel):
blocked_by: str
class Items(BaseModel):
items: List[str]
class ItemPatch(BaseModel):
done: Optional[bool] = None
text: Optional[str] = None
class Similar(BaseModel):
text: Optional[str] = None
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)) from None
except Forbidden as e:
raise HTTPException(403, str(e)) from None
except Conflict as e:
raise HTTPException(409, str(e)) from None
except Invalid as e:
raise HTTPException(422, str(e)) from None
# who am I -----------------------------------------------------------------------------
@r.get("/api/me")
def whoami(request: Request):
return {"email": me(request).email}
# boards --------------------------------------------------------------------------------
@r.get("/api/boards")
def my_boards(request: Request):
return run(boards.first_visit, me(request))
@r.post("/api/boards", status_code=201)
def create_board(body: NewBoard, request: Request):
return run(boards.create_board, me(request), body.title, None, body.template)
@r.get("/api/templates")
def templates(request: Request):
me(request)
return boards.templates()
@r.get("/api/boards/{board_id}/stamp")
def stamp(board_id: str, request: Request):
return run(boards.board_stamp, me(request), board_id)
@r.get("/api/boards/{board_id}/archived")
def archived(board_id: str, request: Request):
return run(boards.archived_cards, me(request), board_id)
@r.get("/api/boards/{board_id}")
def show_board(board_id: str, request: Request):
return run(boards.show_board, me(request), board_id)
@r.patch("/api/boards/{board_id}")
def rename_board(board_id: str, body: Title, request: Request):
return run(boards.rename_board, me(request), board_id, body.title)
@r.get("/api/boards/{board_id}/activity")
def activity(board_id: str, request: Request):
return run(boards.board_activity, me(request), board_id)
@r.get("/api/boards/{board_id}/summary")
def summary(board_id: str, request: Request):
return run(boards.daily_summary, me(request), board_id)
@r.put("/api/boards/{board_id}/members", status_code=201)
def invite(board_id: str, body: Member, request: Request):
return run(boards.invite, me(request), board_id, body.email, body.role)
@r.delete("/api/boards/{board_id}/members/{email}")
def remove_member(board_id: str, email: str, request: Request):
return run(boards.remove_member, me(request), board_id, email)
# lists ---------------------------------------------------------------------------------
@r.post("/api/boards/{board_id}/lists", status_code=201)
def add_list(board_id: str, body: ListIn, request: Request):
return run(boards.add_list, me(request), board_id, body.name, body.is_done)
@r.patch("/api/boards/{board_id}/lists/{list_id}")
def update_list(board_id: str, list_id: str, body: ListPatch, request: Request):
return run(boards.update_list, me(request), board_id, list_id, body.name,
body.is_done, body.position)
# cards ---------------------------------------------------------------------------------
@r.post("/api/boards/{board_id}/cards", status_code=201)
def add_card(board_id: str, body: CardIn, request: Request):
return run(boards.add_card, me(request), board_id, body.title, body.list_name,
body.description, body.assignee, body.due_date, body.labels)
@r.get("/api/cards/{card_id}")
def card_detail(card_id: str, request: Request):
return run(boards.card_detail, me(request), card_id)
@r.patch("/api/cards/{card_id}")
def update_card(card_id: str, body: CardPatch, request: Request):
kwargs = {"title": body.title, "description": body.description, "labels": body.labels}
if body.assignee != "__unset__":
kwargs["assignee"] = body.assignee
if body.due_date != "__unset__":
kwargs["due_date"] = body.due_date
return run(boards.update_card, me(request), card_id, None, body.expected_version,
**kwargs)
@r.post("/api/cards/{card_id}/move")
def move_card(card_id: str, body: Move, request: Request):
return run(boards.move_card, me(request), card_id, body.list_name, None, body.before_card,
body.expected_version)
@r.get("/api/cards/{card_id}/history")
def card_history(card_id: str, request: Request):
return run(boards.card_history, me(request), card_id)
@r.post("/api/cards/{card_id}/blockers", status_code=201)
def add_blocker(card_id: str, body: Blocker, request: Request):
return run(boards.add_blocker, me(request), card_id, body.blocked_by)
@r.delete("/api/cards/{card_id}/blockers/{blocker_id}")
def remove_blocker(card_id: str, blocker_id: str, request: Request):
return run(boards.remove_blocker, me(request), card_id, blocker_id)
@r.get("/api/cards/{card_id}/blocking")
def what_is_blocking(card_id: str, request: Request):
return run(boards.what_is_blocking, me(request), card_id)
@r.post("/api/cards/{card_id}/checklist/suggest")
def suggest_checklist(card_id: str, request: Request):
return run(boards.suggest_checklist, me(request), card_id)
@r.post("/api/cards/{card_id}/checklist", status_code=201)
def add_items(card_id: str, body: Items, request: Request):
return run(boards.add_checklist_items, me(request), card_id, body.items)
@r.patch("/api/cards/{card_id}/checklist/{item_id}")
def set_item(card_id: str, item_id: str, body: ItemPatch, request: Request):
return run(boards.set_checklist_item, me(request), card_id, item_id, body.done,
body.text)
@r.delete("/api/cards/{card_id}/checklist/{item_id}")
def delete_item(card_id: str, item_id: str, request: Request):
return run(boards.set_checklist_item, me(request), card_id, item_id, remove=True)
@r.post("/api/cards/{card_id}/similar")
def similar(card_id: str, request: Request):
return run(boards.similar_cards, me(request), card_id)
@r.post("/api/cards/{card_id}/comments", status_code=201)
def add_comment(card_id: str, body: CommentIn, request: Request):
return run(boards.add_comment, me(request), card_id, body.body)
@r.get("/api/cards/{card_id}/comments")
def comments(card_id: str, request: Request):
return run(boards.comments, me(request), card_id)
@r.post("/api/cards/{card_id}/archive")
def archive(card_id: str, request: Request):
return run(boards.archive_card, me(request), card_id)
@r.post("/api/cards/{card_id}/restore")
def restore(card_id: str, request: Request):
return run(boards.restore_card, me(request), card_id)
# attachments: the raw file is the request body, its name a query parameter (no multipart)
@r.post("/api/cards/{card_id}/attachments", status_code=201)
async def attach(card_id: str, name: str, request: Request):
person = me(request)
if int(request.headers.get("content-length") or 0) > MAX_ATTACHMENT:
raise HTTPException(413, "a file can be at most 10 MB")
data = await request.body()
return run(boards.add_attachment, person, card_id, name, data,
request.headers.get("content-type", ""))
@r.get("/api/attachments/{att_id}")
def download(att_id: str, request: Request):
f = run(boards.download_attachment, me(request), att_id)
how = "inline" if f["inline"] else "attachment"
return Response(f["data"], media_type=f["content_type"], headers={
"Content-Disposition": f"{how}; filename*=UTF-8''{quote(f['name'])}",
"X-Content-Type-Options": "nosniff", "Cache-Control": "private, no-store",
"Content-Security-Policy": "default-src 'none'; img-src 'self'; sandbox"})
@r.delete("/api/attachments/{att_id}")
def delete_attachment(att_id: str, request: Request):
return run(boards.delete_attachment, me(request), att_id)
# across boards -------------------------------------------------------------------------
@r.get("/api/me/notifications")
def notifications(request: Request, unread: bool = False):
return run(boards.my_notifications, me(request), unread)
@r.post("/api/me/notifications/read")
def mark_read(body: ReadIn, request: Request):
return run(boards.mark_notifications_read, me(request), body.ids)
@r.get("/api/me/week")
def my_week(request: Request, days: int = 7):
return run(boards.my_cards, me(request), days)
@r.get("/api/overdue")
def overdue(request: Request):
return run(boards.overdue, me(request))
@r.get("/api/search")
def search(request: Request, q: str):
return run(boards.search_cards, me(request), q)
@r.post("/api/similar")
def similar_text(body: Similar, request: Request):
return run(boards.similar_cards, me(request), None, body.text)
# the page ------------------------------------------------------------------------------
@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 FileResponse(STATIC / "index.html", headers={"Cache-Control": "no-cache"})
@r.get("/static/{name}")
def static(name: str):
f = (STATIC / name).resolve()
if f.parent != STATIC.resolve() or not f.is_file():
raise HTTPException(404, "not found")
return FileResponse(f, headers={"Cache-Control": "no-cache"})
return r