dmslanes/lanes

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

• dodil-challenge
• dodil-challenge-boards

A remix template · 0 remixes · Remix dmslanes/lanes

app/infrastructure/memory.py

"""In-memory Repository: the tests and `LANES_STORAGE=memory` local runs use it."""
from __future__ import annotations

import threading
from dataclasses import replace

from app.domain.model import Activity, Board, BoardList, Card, Conflict, Membership, Workspace


class MemoryRepository:
    def __init__(self):
        self._lock = threading.Lock()
        self.workspaces: dict[str, Workspace] = {}
        self.members: dict[tuple[str, str], Membership] = {}
        self.boards: dict[str, Board] = {}
        self.lists: dict[str, BoardList] = {}
        self.cards: dict[str, Card] = {}
        self.activity: list[Activity] = []
        self._card_number = 0

    def add_workspace(self, ws):
        self.workspaces[ws.id] = ws

    def get_workspace(self, workspace_id):
        return self.workspaces.get(workspace_id)

    def get_membership(self, workspace_id, email):
        return self.members.get((workspace_id, email))

    def memberships_of_workspace(self, workspace_id):
        return [m for (w, _), m in self.members.items() if w == workspace_id]

    def memberships_of_person(self, email):
        return [m for (_, e), m in self.members.items() if e == email]

    def save_membership(self, m):
        self.members[(m.workspace_id, m.email)] = m

    def delete_membership(self, workspace_id, email):
        self.members.pop((workspace_id, email), None)

    def add_board(self, board):
        self.boards[board.id] = board

    def get_board(self, board_id):
        return self.boards.get(board_id)

    def boards_of_workspaces(self, workspace_ids):
        ids = set(workspace_ids)
        return [b for b in self.boards.values() if b.workspace_id in ids]

    def add_list(self, lst):
        self.lists[lst.id] = lst

    def get_list(self, list_id):
        return self.lists.get(list_id)

    def lists_of_board(self, board_id):
        return sorted([l for l in self.lists.values() if l.board_id == board_id], key=lambda l: l.position)

    def claim_card_number(self):
        with self._lock:
            self._card_number += 1
            return self._card_number

    def add_card(self, card):
        self.cards[card.id] = replace(card)

    def save_card(self, card, expected_version):
        with self._lock:
            current = self.cards.get(card.id)
            if current is None or current.version != expected_version:
                raise Conflict("this card changed since you opened it; reload and try again")
            self.cards[card.id] = replace(card)

    def delete_card(self, card_id):
        self.cards.pop(card_id, None)

    def get_card(self, card_id):
        c = self.cards.get(card_id)
        return replace(c) if c else None

    def get_card_by_number(self, number):
        for c in self.cards.values():
            if c.number == number:
                return replace(c)
        return None

    def cards_of_board(self, board_id):
        return [replace(c) for c in self.cards.values() if c.board_id == board_id]

    def cards_of_workspaces(self, workspace_ids):
        ids = set(workspace_ids)
        return [replace(c) for c in self.cards.values() if c.workspace_id in ids]

    def add_activity(self, a):
        self.activity.append(a)

    def activity_of_card(self, card_id):
        return [a for a in self.activity if a.card_id == card_id]

    def activity_of_board(self, board_id, limit):
        return sorted([a for a in self.activity if a.board_id == board_id], key=lambda a: a.at, reverse=True)[:limit]