flc/crewboard

Crewboard - team task board (boards, lists, cards, roles, history, drag and drop, MCP, AI extras). FastAPI + DataK3 + appid + Models.

• dodil-challenge
• dodil-challenge-boards

A remix template · 1 remixes · Remix flc/crewboard

app/boards/infrastructure/memory.py

"""An in-memory repository: the tests use it, and so does `DEV_USER=... uvicorn` on localhost when
no BUCKET is set, so the whole board runs without a Dodil account."""
import copy
from datetime import datetime, timezone

from boards.application.ports import Taken


class MemoryBoards:
    """Stores copies, like a database would, so a use case can't cheat by mutating shared objects."""

    def __init__(self, clock=None):
        self.boards, self.cards, self.events = {}, {}, []
        self.clock = clock or (lambda: datetime.now(timezone.utc).isoformat(timespec="seconds"))

    def get(self, board_id):
        return copy.deepcopy(self.boards.get(board_id))

    def boards_for(self, email):
        return [copy.deepcopy(b) for b in self.boards.values()
                if b.owner_email == email or email in b.members]

    def insert_board(self, board):
        if board.id in self.boards:
            raise Taken(board.id)
        self.boards[board.id] = copy.deepcopy(board)

    def save_board(self, board):
        stored = self.boards[board.id]
        stored.title, stored.lists = board.title, list(board.lists)

    def save_members(self, board, removed=()):
        self.boards[board.id].members = dict(board.members)

    def get_card(self, card_id):
        return copy.deepcopy(self.cards.get(card_id))

    def cards_on(self, board_id):
        return sorted((copy.deepcopy(c) for c in self.cards.values() if c.board_id == board_id),
                      key=lambda c: c.position)

    def cards_on_boards(self, board_ids):
        return [c for b in board_ids for c in self.cards_on(b)]

    def max_number(self, board_id):
        return max((c.number for c in self.cards.values() if c.board_id == board_id), default=0)

    def insert_card(self, card):
        if card.id in self.cards:
            raise Taken(card.id)
        self.cards[card.id] = copy.deepcopy(card)

    def update_card(self, card, expected_version):
        if self.cards[card.id].version != expected_version:
            return False
        self.cards[card.id] = copy.deepcopy(card)
        return True

    def log(self, board_id, card_id, actor, action, changes=None):
        self.events.append({"board_id": board_id, "card_id": card_id, "actor": actor,
                            "action": action, "changes": changes,
                            "at": self.clock()})

    def activity(self, board_id, card_id=None, limit=50):
        rows = [e for e in self.events if e["board_id"] == board_id
                and (card_id is None or e["card_id"] == card_id)]
        return list(reversed(rows))[:limit]