contest/orbit-boards

Orbit Boards, a private, remixable team task board built with FastAPI, DataK3, AppID, Models and MCP.

• dodil-challenge
• dodil-challenge-boards

A remix template · 0 remixes · Remix contest/orbit-boards

app/boards/doors/web.py

"""HTTP door. Routes validate transport input and delegate all rules to application use cases."""
import logging
import uuid
from pathlib import Path

import httpx
import psycopg
from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from pydantic import BaseModel, ConfigDict, Field

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 = Path(__file__).parent.parent / "static" / "index.html"
logger = logging.getLogger(__name__)


def build_router(boards: Boards) -> APIRouter:
    router = APIRouter()

    def me(request: Request) -> Person:
        person = person_from(request.headers)
        if person is None:
            raise HTTPException(401, "sign in first: /.dodil/auth/login")
        return person

    def run(fn, *args, **kwargs):
        try:
            return fn(*args, **kwargs)
        except NotFound as error:
            raise HTTPException(404, str(error))
        except Forbidden as error:
            raise HTTPException(403, str(error))
        except Invalid as error:
            raise HTTPException(422, str(error))
        except ValueError as error:
            raise HTTPException(422, "invalid value") from error
        except (psycopg.Error, httpx.HTTPError) as error:
            reference = uuid.uuid4().hex[:10]
            logger.exception("External service request failed (reference=%s)", reference)
            raise HTTPException(
                503, f"A required service is temporarily unavailable. Retry shortly (reference {reference})."
            ) from error
        except Exception as error:
            reference = uuid.uuid4().hex[:10]
            logger.exception("Unhandled request failure (reference=%s)", reference)
            raise HTTPException(
                500, f"The request could not be completed. Please try again (reference {reference})."
            ) from error

    class StrictModel(BaseModel):
        model_config = ConfigDict(extra="forbid", str_strip_whitespace=False)

    class Title(StrictModel):
        title: str = Field(min_length=1, max_length=120)

    class CardCreate(StrictModel):
        title: str = Field(min_length=1, max_length=240)
        list_name: str = "To do"
        description: str = Field(default="", max_length=10000)
        assignee: str | None = None
        due_date: str | None = None
        labels: list[str] = Field(default_factory=list, max_length=12)

    class Move(StrictModel):
        list_name: str
        position: int | None = Field(default=None, ge=0)

    class CardUpdate(StrictModel):
        title: str | None = Field(default=None, min_length=1, max_length=240)
        description: str | None = Field(default=None, max_length=10000)
        assignee: str | None = None
        due_date: str | None = None
        labels: list[str] | None = Field(default=None, max_length=12)

    class Member(StrictModel):
        email: str = Field(min_length=3, max_length=254)
        role: str = "member"

    @router.get("/api/boards")
    def my_boards(request: Request):
        return run(boards.my_boards, me(request))

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

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

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

    @router.get("/api/boards/{board_id}/activity")
    def get_activity(board_id: str, request: Request, limit: int = 50):
        return run(boards.get_activity, me(request), board_id, limit)

    @router.post("/api/boards/{board_id}/cards", status_code=201)
    def add_card(board_id: str, body: CardCreate, request: Request):
        return run(boards.add_card, me(request), board_id, **body.model_dump())

    @router.patch("/api/cards/{card_id}")
    def update_card(card_id: str, body: CardUpdate, request: Request):
        changes = body.model_dump(exclude_unset=True)
        return run(boards.update_card, me(request), card_id, **changes)

    @router.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, body.position)

    @router.post("/api/cards/{card_id}/assign")
    def assign_card(card_id: str, body: Member, request: Request):
        return run(boards.assign_card, me(request), card_id, body.email)

    @router.delete("/api/cards/{card_id}/assignee")
    def unassign_card(card_id: str, request: Request):
        return run(boards.assign_card, me(request), card_id, None)

    @router.get("/api/my-cards")
    def my_cards(request: Request, days: int = 7):
        return run(boards.list_my_cards, me(request), days)

    @router.get("/api/overdue")
    def overdue(request: Request, board_id: str | None = None):
        return run(boards.list_overdue, me(request), board_id)

    @router.post("/api/cards/{card_id}/checklist")
    def suggest_checklist(card_id: str, request: Request):
        return run(boards.suggest_checklist, me(request), card_id)

    @router.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 HTMLResponse(PAGE_PATH.read_text(encoding="utf-8"))

    return router