dmslanes/lanes

Lanes - team task boards with roles, history and an MCP endpoint for Claude and ChatGPT

A remix template · 0 remixes · Remix dmslanes/lanes

app/domain/model.py

"""Domain model: workspaces, people, boards, lists, cards and their activity.

Pure Python, no I/O. Every permission rule lives here so the web routes and the
MCP tools cannot disagree about who may do what.
"""
from __future__ import annotations

import re

from dataclasses import dataclass, field
from datetime import date, datetime
from enum import Enum


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


class DomainError(Exception):
    """A rule was broken. `code` maps to an HTTP status in the doors."""

    code = "invalid"


class NotFound(DomainError):
    code = "not_found"


class Forbidden(DomainError):
    code = "forbidden"


class Conflict(DomainError):
    code = "conflict"


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


@dataclass(frozen=True)
class Membership:
    workspace_id: str
    email: str
    role: Role

    def can_edit(self) -> bool:
        return self.role in (Role.OWNER, Role.MEMBER)

    def can_manage_members(self) -> bool:
        return self.role is Role.OWNER

    def require_edit(self) -> None:
        if not self.can_edit():
            raise Forbidden("viewers can read this board but not change it")

    def require_owner(self) -> None:
        if not self.can_manage_members():
            raise Forbidden("only the workspace owner manages members")


@dataclass
class Workspace:
    id: str
    name: str
    created_by: str
    created_at: datetime


@dataclass
class Board:
    id: str
    workspace_id: str
    name: str
    created_at: datetime


@dataclass
class BoardList:
    id: str
    board_id: str
    name: str
    position: int


@dataclass
class Card:
    id: str
    workspace_id: str
    board_id: str
    list_id: str
    number: int  # human handle, unique per workspace ("card 12")
    title: str
    description: str = ""
    assignee: str | None = None
    due_date: date | None = None
    labels: list[str] = field(default_factory=list)
    position: int = 0
    created_at: datetime | None = None
    updated_at: datetime | None = None
    version: int = 1
    blocked_by: list[int] = field(default_factory=list)  # numbers of the cards that must finish first

    def is_overdue(self, today: date, done_list_ids: set[str]) -> bool:
        return self.due_date is not None and self.due_date < today and self.list_id not in done_list_ids


@dataclass(frozen=True)
class Activity:
    id: str
    card_id: str
    board_id: str
    actor: str
    at: datetime
    action: str  # created | updated | moved | assigned | deleted
    changes: dict  # field -> [old, new]


EDITABLE_CARD_FIELDS = ("title", "description", "assignee", "due_date", "labels", "blocked_by")
DONE_LIST_NAMES = {"done", "finished", "complete", "completed", "shipped"}


def clean_title(title: str, what: str = "title") -> str:
    title = (title or "").strip()
    if not title:
        raise DomainError(f"{what} can't be empty")
    if len(title) > 200:
        raise DomainError(f"{what} is longer than 200 characters")
    return title


def clean_labels(labels: list[str] | None) -> list[str]:
    seen: list[str] = []
    for label in labels or []:
        label = label.strip().lower()[:30]
        if label and label not in seen:
            seen.append(label)
    return seen[:10]


def would_cycle(card_number: int, blocker_number: int, blocked_by_of) -> bool:
    """True if `card` waiting on `blocker` closes a loop (blocker already waits on card, at any depth)."""
    seen, todo = set(), [blocker_number]
    while todo:
        n = todo.pop()
        if n == card_number:
            return True
        if n not in seen:
            seen.add(n)
            todo.extend(blocked_by_of(n))
    return False


def parse_checklist(text: str, limit: int = 10) -> list[str]:
    """Steps out of a model reply: one per line, bullets and numbering stripped."""
    steps = []
    for line in (text or "").splitlines():
        line = re.sub(r"^\s*(?:[-*\u2022]|\d+[.)]|\[[ x]\])\s*(?:\[[ x]\]\s*)?", "", line).strip()
        if line and len(line) <= 200 and line not in steps:
            steps.append(line)
    return steps[:limit]


def diff_card(before: Card, after: Card) -> dict:
    """field -> [old, new] for every editable field that changed (JSON-safe)."""
    out = {}
    for name in EDITABLE_CARD_FIELDS + ("list_id", "position"):
        old, new = getattr(before, name), getattr(after, name)
        if old != new:
            out[name] = [_json(old), _json(new)]
    return out


def _json(value):
    if isinstance(value, (date, datetime)):
        return value.isoformat()
    return value