contest/orbit-boards
Orbit Boards, a private, remixable team task board built with FastAPI, DataK3, AppID, Models and MCP.
A remix template · 0 remixes · Remix contest/orbit-boards
app/boards/application/use_cases.py
"""Shared application operations. The web and MCP doors delegate here for identical rules."""
import copy
import uuid
from datetime import date, timedelta
from dataclasses import asdict
from typing import Optional
from boards.application.ports import Assistant, BoardRepository
from boards.domain.model import LISTS, Board, Card, Forbidden, Invalid, Person, Role
class NotFound(Exception):
"""The requested resource does not exist or is not visible to this caller."""
_UNSET = object()
def _card(card: Card) -> dict:
result = asdict(card)
result["due_date"] = card.due_date.isoformat() if card.due_date else None
result["labels"] = list(card.labels)
return result
def _iso(value):
if isinstance(value, (date,)):
return value.isoformat()
return value
class Boards:
def __init__(self, repo: BoardRepository, assistant: Optional[Assistant] = None):
self.repo, self.assistant = repo, assistant
def _board(self, board_id: str, person: Person) -> Board:
board = self.repo.get(board_id)
if board is None:
raise NotFound("no such board")
try:
board.require_view(person)
except Forbidden:
raise NotFound("no such board")
return board
def _find_card(self, card_id: str, person: Person) -> tuple[Board, Card]:
card = self.repo.get_card(card_id)
if card is None:
raise NotFound("no such card")
return self._board(card.board_id, person), card
def _persist_board_event(self, board: Board, event: dict) -> None:
if hasattr(self.repo, "save_board_event"):
self.repo.save_board_event(board, event)
else:
self.repo.save_board(board)
self.repo.log(board.id, None, event["actor"], event["action"])
def _persist_cards(self, cards: list[Card], events: list[dict]) -> None:
if hasattr(self.repo, "save_cards_and_events"):
self.repo.save_cards_and_events(cards, events)
return
for card in cards:
self.repo.save_card(card)
for event in events:
self.repo.log(event["board_id"], event.get("card_id"), event["actor"],
event["action"])
@staticmethod
def _event(board_id: str, card_id: Optional[str], person: Person, action: str,
field: Optional[str] = None, old_value=None, new_value=None) -> dict:
return {"board_id": board_id, "card_id": card_id, "actor": person.email,
"action": action, "field": field, "old_value": _iso(old_value),
"new_value": _iso(new_value)}
def create_board(self, person: Person, title: str) -> dict:
title = title.strip()
if not title:
raise Invalid("a board needs a title")
if len(title) > 120:
raise Invalid("board titles must be 120 characters or fewer")
board = Board(id=uuid.uuid4().hex[:12], title=title, owner_email=person.email)
self._persist_board_event(board, self._event(board.id, None, person,
f"created board {title!r}", "title", None, title))
return {"id": board.id, "title": board.title}
def my_boards(self, person: Person) -> list[dict]:
return [{"id": board.id, "title": board.title, "role": board.role_of(person).value}
for board in self.repo.boards_for(person.email)]
def show_board(self, person: Person, board_id: str) -> dict:
board = self._board(board_id, person)
cards = sorted(self.repo.cards_on(board.id), key=lambda c: (c.list_name, c.position, c.id))
people = [{"email": board.owner_email, "role": Role.OWNER.value}]
people += [{"email": email, "role": role.value} for email, role in sorted(board.members.items())]
return {"id": board.id, "title": board.title, "role": board.role_of(person).value,
"lists": list(LISTS),
"members": people, "cards": [_card(card) for card in cards]}
def add_member(self, person: Person, board_id: str, email: str, role: str = "member") -> dict:
board = self._board(board_id, person)
try:
selected_role = Role(role)
except ValueError:
raise Invalid("role must be member or viewer")
board.add_member(person, email, selected_role)
normalized_email = email.strip().lower()
self._persist_board_event(board, self._event(board.id, None, person,
f"added {normalized_email} as {selected_role.value}", "membership", None,
{"email": normalized_email, "role": selected_role.value}))
return {"board_id": board.id, "email": normalized_email, "role": selected_role.value}
def add_card(self, person: Person, board_id: str, title: str, list_name: str = "To do",
description: str = "", assignee: Optional[str] = None,
due_date: Optional[str | date] = None, labels: Optional[list[str]] = None) -> dict:
board = self._board(board_id, person)
board.require_edit(person)
if assignee and not board.can_assign(assignee):
raise Invalid("assignee must be a member of this board")
parsed_due = date.fromisoformat(due_date) if isinstance(due_date, str) and due_date else due_date
position = len([c for c in self.repo.cards_on(board.id) if c.list_name == list_name])
card = Card(id=uuid.uuid4().hex[:12], board_id=board.id, title=title,
list_name=list_name, description=description, assignee=assignee.strip().lower()
if assignee else None, due_date=parsed_due, labels=labels or [], position=position)
card.validate()
event = self._event(board.id, card.id, person, f"created card {card.title!r}",
"created", None, {"title": card.title, "list_name": card.list_name})
self._persist_cards([card], [event])
return _card(card)
def move_card(self, person: Person, card_id: str, list_name: str,
position: Optional[int] = None) -> dict:
board, card = self._find_card(card_id, person)
board.require_edit(person)
if list_name not in LISTS:
card.move_to(list_name, position)
old_list, old_position = card.list_name, card.position
target = [c for c in self.repo.cards_on(board.id)
if c.list_name == list_name and c.id != card.id]
target.sort(key=lambda c: (c.position, c.id))
index = len(target) if position is None else min(position, len(target))
if index < 0:
raise Invalid("card position must be zero or greater")
target.insert(index, card)
card.move_to(list_name, index)
affected = target
if old_list != list_name:
source = [c for c in self.repo.cards_on(board.id)
if c.list_name == old_list and c.id != card.id]
source.sort(key=lambda c: (c.position, c.id))
for i, item in enumerate(source):
item.position = i
affected += source
for i, item in enumerate(target):
item.position = i
event = self._event(board.id, card.id, person, "moved card", "list_name",
old_list, list_name)
if old_list == list_name and old_position != index:
event["action"] = "reordered card"
event["field"] = "position"
event["old_value"], event["new_value"] = old_position, index
self._persist_cards(affected, [event])
return _card(card)
def update_card(self, person: Person, card_id: str, **changes) -> dict:
board, original = self._find_card(card_id, person)
board.require_edit(person)
card = copy.deepcopy(original)
allowed = {"title", "description", "assignee", "due_date", "labels"}
unknown = set(changes) - allowed
if unknown:
raise Invalid(f"unsupported card field: {sorted(unknown)[0]}")
old_values = {key: getattr(card, key) for key in changes}
for key, value in changes.items():
if key == "due_date":
value = date.fromisoformat(value) if isinstance(value, str) and value else None
elif key == "assignee":
value = value.strip().lower() if isinstance(value, str) and value.strip() else None
if value and not board.can_assign(value):
raise Invalid("assignee must be a member of this board")
elif key == "description":
value = value or ""
elif key == "labels":
if not isinstance(value, list):
raise Invalid("labels must be a list of names")
setattr(card, key, value)
card.validate()
events = []
for key, old in old_values.items():
new = getattr(card, key)
if old != new:
events.append(self._event(board.id, card.id, person, f"changed {key}", key, old, new))
if events:
self._persist_cards([card], events)
return _card(card)
def assign_card(self, person: Person, card_id: str, email: Optional[str]) -> dict:
return self.update_card(person, card_id, assignee=email)
def list_my_cards(self, person: Person, days: int = 7) -> list[dict]:
if days < 0 or days > 365:
raise Invalid("days must be between 0 and 365")
end = date.today() + timedelta(days=days)
result = []
for board in self.repo.boards_for(person.email):
for card in self.repo.cards_on(board.id):
if card.assignee == person.email and card.list_name != "Done" and card.due_date:
if card.due_date <= end:
result.append(_card(card))
return sorted(result, key=lambda c: (c["due_date"] or "9999-12-31", c["position"], c["id"]))
def list_overdue(self, person: Person, board_id: Optional[str] = None) -> list[dict]:
boards = [self._board(board_id, person)] if board_id else self.repo.boards_for(person.email)
today = date.today()
result = []
for board in boards:
for card in self.repo.cards_on(board.id):
if card.due_date is not None and card.due_date < today and card.list_name != "Done":
result.append(_card(card))
return sorted(result, key=lambda c: (c["due_date"], c["id"]))
def get_activity(self, person: Person, board_id: str, limit: int = 50) -> list[dict]:
board = self._board(board_id, person)
if limit < 1 or limit > 100:
raise Invalid("activity limit must be between 1 and 100")
if hasattr(self.repo, "activity_for"):
return self.repo.activity_for(board.id, limit)
return []
def suggest_checklist(self, person: Person, card_id: str) -> dict:
_, card = self._find_card(card_id, person)
if self.assistant is None:
raise Invalid("no AI assistant configured")
return {"card_id": card.id, "suggested_checklist": self.assistant.checklist(card.title)}