rgboards/rg-boards

Lanes, team task boards you run from the web or from Claude/ChatGPT (MCP). FastAPI + DataK3 + appid + Dodil Models.

A remix template · 0 remixes · Remix rgboards/rg-boards

app/boards/application/ports.py

"""What the use cases need from the outside world. infrastructure/ provides it; tests fake it.

Two implementations of BoardRepository ship with the app:
  infrastructure/repository.py  DataK3 (production)
  infrastructure/memory.py      in memory (tests, and local runs without a Dodil account)
"""
from datetime import datetime
from typing import Optional, Protocol

from boards.domain.model import Board, Card


class DuplicateKey(Exception):
    """A plain INSERT lost a race for its key (Postgres 23505)."""


class BoardRepository(Protocol):
    # boards, members and lists
    def get(self, board_id: str) -> Optional[Board]: ...
    def boards_for(self, email: str) -> list: ...
    def save_board(self, board: Board) -> None: ...

    # cards
    def get_card(self, card_id: str) -> Optional[Card]: ...
    def cards_on(self, board_ids: list) -> list: ...
    def next_number(self, board_id: str) -> int: ...
    def insert_card(self, card: Card) -> None: ...          # plain INSERT: DuplicateKey on a race
    def update_card(self, card: Card, expected_version: int) -> bool: ...  # False if stale

    # history: who changed what, when
    def log(self, board_id: str, card_id: Optional[str], actor: str, action: str,
            field: Optional[str] = None, old: Optional[str] = None,
            new: Optional[str] = None, at: Optional[datetime] = None) -> None: ...
    def history(self, card_id: str) -> list: ...
    def activity(self, board_id: str, limit: int = 50) -> list: ...

    # "blocked by" links: (card_id, blocked_by_id)
    def links_on(self, board_id: str) -> list: ...
    def add_link(self, board_id: str, card_id: str, blocked_by: str, actor: str) -> None: ...
    def remove_link(self, card_id: str, blocked_by: str) -> None: ...

    # checklists: dicts {id, card_id, text, done, position}
    def checklist(self, card_id: str) -> list: ...
    def save_check_item(self, board_id: str, item: dict) -> None: ...

    # semantic search cache: card_id -> (text_hash, vector)
    def vectors_on(self, board_ids: list) -> dict: ...
    def save_vector(self, board_id: str, card_id: str, text_hash: str, vector: list) -> None: ...

    # the day's AI summary of a board: {"summary": str, "written_at": str} or None
    def get_summary(self, board_id: str, day: str) -> Optional[dict]: ...
    def save_summary(self, board_id: str, day: str, text: str) -> None: ...


class Assistant(Protocol):
    """Dodil Models behind a small, task-shaped interface."""
    def checklist(self, card_title: str, description: str = "") -> list: ...
    def summary(self, board_title: str, facts: str) -> str: ...
    def embed(self, texts: list) -> list: ...