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/domain/model.py

"""TaskFlow's language and rules. No database, no HTTP: plain Python that tests in milliseconds.

Every rule lives here, so it holds the same way for the web page and for Claude or ChatGPT.
"""
from dataclasses import dataclass, field
from datetime import date
from enum import Enum
from typing import Optional

DEFAULT_LISTS = ("To do", "Doing", "Done")
DONE_NAMES = ("done", "completed", "finished", "closed")
MAX_LISTS = 12
MAX_LABELS = 10
MAX_CARDS = 1000            # per board: a cap, so a script can't flood a board
EDITABLE = ("title", "description", "assignee", "due_date", "labels")


class Role(str, Enum):
    OWNER = "owner"
    MEMBER = "member"
    VIEWER = "viewer"


class Forbidden(Exception):
    """The person may not do this. The doors turn it into 403 (web) or a tool error (MCP)."""


class Invalid(Exception):
    """The request breaks a rule of the board itself. 422 (web) or a tool error (MCP)."""


def clean_email(email: str) -> str:
    email = (email or "").strip().lower()
    if len(email) > 254 or "@" not in email or email.startswith("@") or email.endswith("@") or " " in email:
        raise Invalid(f"{email!r} is not an email address")
    return email


def clean_labels(labels) -> tuple:
    out = []
    for label in labels or ():
        label = str(label).strip().lower()
        if label and label not in out:
            out.append(label[:30])
    if len(out) > MAX_LABELS:
        raise Invalid(f"a card has at most {MAX_LABELS} labels")
    return tuple(out)


@dataclass(frozen=True)
class Person:
    """Who is acting: taken from the gateway's sign-in, never from the request body."""
    email: str


@dataclass
class Board:
    id: str
    title: str
    owner_email: str
    members: dict = field(default_factory=dict)        # email -> Role, owner not included
    lists: list = field(default_factory=lambda: list(DEFAULT_LISTS))
    card_seq: int = 0                                   # highest card number claimed so far

    # --- who may do what -------------------------------------------------------------------

    def role_of(self, person: Person) -> Optional[Role]:
        if person.email == self.owner_email:
            return Role.OWNER
        return self.members.get(person.email)

    def require_view(self, person: Person) -> Role:
        role = self.role_of(person)
        if role is None:
            raise Forbidden("you are not on this board")
        return role

    def require_edit(self, person: Person) -> None:
        if self.require_view(person) is Role.VIEWER:
            raise Forbidden("viewers can't change the board")

    def require_owner(self, person: Person) -> None:
        if self.require_view(person) is not Role.OWNER:
            raise Forbidden("only the owner manages members and lists")

    # --- members ---------------------------------------------------------------------------

    def add_member(self, by: Person, email: str, role: Role) -> str:
        self.require_owner(by)
        email = clean_email(email)
        if role is Role.OWNER:
            raise Invalid("a board has exactly one owner")
        if email == self.owner_email:
            raise Invalid("that person already owns this board")
        self.members[email] = role
        return email

    def remove_member(self, by: Person, email: str) -> str:
        self.require_owner(by)
        email = clean_email(email)
        if email == self.owner_email:
            raise Invalid("the owner can't be removed")
        if email not in self.members:
            raise Invalid(f"{email} is not on this board")
        del self.members[email]
        return email

    def can_be_assigned(self, email: str) -> bool:
        """Only people who can work the board (owner, members) can hold a card."""
        return email == self.owner_email or self.members.get(email) is Role.MEMBER

    # --- lists -----------------------------------------------------------------------------

    @property
    def done_list(self) -> str:
        """The list that means finished work: its cards are never overdue or blocking. A list
        called Done (or Completed, Finished, Closed) wherever it sits, else the last list."""
        for name in self.lists:
            if name.lower() in DONE_NAMES:
                return name
        return self.lists[-1]

    def list_named(self, name: str) -> str:
        """Lists are matched without caring about case, so 'doing' finds 'Doing'."""
        for existing in self.lists:
            if existing.lower() == (name or "").strip().lower():
                return existing
        raise Invalid(f"no list {name!r}; lists are {', '.join(self.lists)}")

    def add_list(self, by: Person, name: str, before: Optional[str] = None) -> str:
        self.require_owner(by)
        name = (name or "").strip()
        if not name:
            raise Invalid("a list needs a name")
        if any(existing.lower() == name.lower() for existing in self.lists):
            raise Invalid(f"there is already a list called {name!r}")
        if len(self.lists) >= MAX_LISTS:
            raise Invalid(f"a board has at most {MAX_LISTS} lists")
        if before:
            at = self.lists.index(self.list_named(before))
        elif self.lists and self.done_list.lower() in DONE_NAMES:
            at = self.lists.index(self.done_list)        # new work goes before Done by default
        else:
            at = len(self.lists)
        self.lists.insert(at, name[:40])
        return name

    def rename_list(self, by: Person, old: str, new: str) -> tuple:
        self.require_owner(by)
        old, new = self.list_named(old), (new or "").strip()[:40]
        if not new:
            raise Invalid("a list needs a name")
        if new.lower() != old.lower() and any(e.lower() == new.lower() for e in self.lists):
            raise Invalid(f"there is already a list called {new!r}")
        self.lists[self.lists.index(old)] = new
        return old, new

    def remove_list(self, by: Person, name: str, card_count: int) -> str:
        """Only an empty list goes, so deleting a list never deletes work."""
        self.require_owner(by)
        name = self.list_named(name)
        if card_count:
            raise Invalid(f"{name!r} still has {card_count} card(s); move them out first")
        if len(self.lists) == 1:
            raise Invalid("a board needs at least one list")
        self.lists.remove(name)
        return name



@dataclass
class Card:
    id: str
    board_id: str
    number: int                     # #12 on its board: what people (and Claude) call it
    title: str
    list_name: str
    position: float = 0.0           # order inside the list
    description: str = ""
    assignee: Optional[str] = None
    due_date: Optional[date] = None
    labels: tuple = ()
    blocked_by: list = field(default_factory=list)      # ids of cards on the same board
    checklist: list = field(default_factory=list)       # [{"text": str, "done": bool}]

    def edit(self, board: Board, **changes) -> dict:
        """Apply the changes that really change something. Returns {field: [before, after]}
        so the activity log can say exactly what happened."""
        diff = {}
        for name, value in changes.items():
            if name not in EDITABLE:
                raise Invalid(f"{name!r} can't be edited")
            if name == "title":
                value = (value or "").strip()
                if not value:
                    raise Invalid("a card needs a title")
                value = value[:200]
            elif name == "description":
                value = (value or "").strip()[:5000]
            elif name == "assignee":
                value = clean_email(value) if value else None
                if value and not board.can_be_assigned(value):
                    raise Invalid(f"{value} can't be assigned: only the owner and members of "
                                  f"this board can hold a card")
            elif name == "due_date":
                value = parse_date(value)
            elif name == "labels":
                value = clean_labels(value)
            before = getattr(self, name)
            if before != value:
                setattr(self, name, value)
                diff[name] = [_plain(before), _plain(value)]
        return diff

    def move_to(self, board: Board, list_name: str, position: float) -> Optional[list]:
        target = board.list_named(list_name)
        before = self.list_name
        self.list_name, self.position = target, position
        return [before, target] if before != target else None

    def is_done(self, board: Board) -> bool:
        return self.list_name == board.done_list

    def is_overdue(self, board: Board, today: date) -> bool:
        return bool(self.due_date and self.due_date < today and not self.is_done(board))

    # --- checklist -------------------------------------------------------------------------

    def add_checklist_items(self, items) -> list:
        added = [str(i).strip()[:200] for i in items or () if str(i).strip()]
        if not added:
            raise Invalid("give at least one checklist item")
        if len(self.checklist) + len(added) > 50:
            raise Invalid("a checklist has at most 50 items")
        self.checklist.extend({"text": text, "done": False} for text in added)
        return added

    def tick(self, index: int, done: bool) -> dict:
        if not 1 <= index <= len(self.checklist):
            raise Invalid(f"there is no checklist item {index}; items are 1 to "
                          f"{len(self.checklist)}")
        item = self.checklist[index - 1]
        item["done"] = bool(done)
        return item


def parse_date(value) -> Optional[date]:
    if value in (None, ""):
        return None
    if isinstance(value, date):
        return value
    try:
        return date.fromisoformat(str(value).strip()[:10])
    except ValueError:
        raise Invalid(f"{value!r} is not a date; use YYYY-MM-DD")


def would_cycle(card_id: str, blocker_id: str, blockers_of) -> bool:
    """True if `card_id` blocked by `blocker_id` closes a loop. `blockers_of(id)` gives the ids
    that block a card; we walk from the blocker and look for the card."""
    seen, todo = set(), [blocker_id]
    while todo:
        current = todo.pop()
        if current == card_id:
            return True
        if current not in seen:
            seen.add(current)
            todo.extend(blockers_of(current))
    return False


def _plain(value):
    if isinstance(value, date):
        return value.isoformat()
    if isinstance(value, tuple):
        return list(value)
    return value