eduard/cobra-boards
Cobra Boards, team task boards you can run from Claude or ChatGPT. FastAPI + DataK3 + appid + Dodil Models + MCP.
A remix template · 1 remixes · Remix eduard/cobra-boards
app/boards/infrastructure/memory.py
"""An in-memory repository with the same contract as the DataK3 one. The tests run on it, and so
does the local preview (`COBRA_MEMORY=1`). Production data always lives in DataK3."""
import copy
import math
from typing import Optional
from boards.domain.model import Board, BoardList, Card, Event, Role
class MemoryBoards:
def __init__(self):
self.boards, self.cards, self.events = {}, {}, []
self.deps, self.checklists, self.vectors = set(), {}, {}
self.comment_rows, self.notes, self.atts, self.archived = [], [], {}, {}
# boards ---------------------------------------------------------------------------------
def get(self, board_id: str) -> Optional[Board]:
b = self.boards.get(board_id)
if b is None:
return None
b = copy.deepcopy(b)
b.lists.sort(key=lambda l: l.position)
return b
def boards_for(self, email: str) -> list:
return [self.get(b.id) for b in self.boards.values()
if b.owner_email == email or email in b.members]
def create_board(self, board: Board) -> None:
if board.id in self.boards: # same contract as the SQL primary key
raise KeyError(f"board {board.id} already exists")
self.boards[board.id] = copy.deepcopy(board)
def rename_board(self, board_id: str, title: str) -> None:
self.boards[board_id].title = title
def save_member(self, board_id: str, email: str, role: str) -> None:
self.boards[board_id].members[email] = Role(role)
def delete_member(self, board_id: str, email: str) -> None:
self.boards[board_id].members.pop(email, None)
def save_list(self, lst: BoardList) -> None:
b = self.boards[lst.board_id]
b.lists = [l for l in b.lists if l.id != lst.id] + [copy.deepcopy(lst)]
# cards ----------------------------------------------------------------------------------
def get_card(self, card_id: str) -> Optional[Card]:
c = self.cards.get(card_id)
return copy.deepcopy(c) if c else None
def card_by_number(self, board_id: str, number: int) -> Optional[Card]:
return self.get_card(f"{board_id}-{int(number)}")
def cards_on(self, board_id: str) -> list:
return sorted((copy.deepcopy(c) for c in self.cards.values() if c.board_id == board_id),
key=lambda c: (c.position, c.number))
def cards_assigned_to(self, email: str) -> list:
return [copy.deepcopy(c) for c in self.cards.values() if c.assignee == email]
def next_number(self, board_id: str) -> int:
return 1 + max((c.number for c in self.cards.values() if c.board_id == board_id),
default=0)
def insert_card(self, card: Card, event: Event) -> bool:
if card.id in self.cards: # the plain INSERT that claims a number: second writer loses
return False
self.cards[card.id] = copy.deepcopy(card)
self.events.append(copy.deepcopy(event))
return True
def update_card(self, card: Card, expected_version: int, event: Event) -> bool:
cur = self.cards.get(card.id)
if cur is None or cur.version != expected_version:
return False
self.cards[card.id] = copy.deepcopy(card)
self.events.append(copy.deepcopy(event))
return True
def unassign_all(self, board_id: str, email: str, event_for) -> int:
n = 0
for c in self.cards.values():
if c.board_id == board_id and c.assignee == email:
c.assignee, c.version = None, c.version + 1
self.events.append(event_for(c))
n += 1
return n
# history --------------------------------------------------------------------------------
def log(self, event: Event) -> None:
self.events.append(copy.deepcopy(event))
def history(self, card_id: str, limit: int = 50) -> list:
return [e for e in reversed(self.events) if e.card_id == card_id][:limit]
def board_activity(self, board_id: str, limit: int = 50) -> list:
return [e for e in reversed(self.events) if e.board_id == board_id][:limit]
# dependencies ---------------------------------------------------------------------------
def blockers_on(self, board_id: str) -> dict:
out = {}
for b, card, blocker in self.deps:
if b == board_id:
out.setdefault(card, set()).add(blocker)
return out
def add_dependency(self, board_id: str, card_id: str, blocker_id: str) -> None:
self.deps.add((board_id, card_id, blocker_id))
def remove_dependency(self, card_id: str, blocker_id: str) -> None:
self.deps = {d for d in self.deps if not (d[1] == card_id and d[2] == blocker_id)}
# checklists -----------------------------------------------------------------------------
def checklist(self, card_id: str) -> list:
items = self.checklists.get(card_id, {})
return sorted((dict(v) for v in items.values()), key=lambda i: i["position"])
def save_checklist_item(self, card_id, item_id, text, done, position) -> None:
self.checklists.setdefault(card_id, {})[item_id] = {
"id": item_id, "text": text, "done": bool(done), "position": float(position)}
def delete_checklist_item(self, card_id, item_id) -> None:
self.checklists.get(card_id, {}).pop(item_id, None)
# meaning search -------------------------------------------------------------------------
def save_embedding(self, card_id, board_id, vector, text_hash) -> None:
self.vectors[card_id] = (board_id, list(vector), text_hash)
def embedding_hash(self, card_id) -> Optional[str]:
v = self.vectors.get(card_id)
return v[2] if v else None
def similar_cards(self, board_ids, vector, limit) -> list:
def cos_dist(a, b):
dot = sum(x * y for x, y in zip(a, b))
na, nb = math.sqrt(sum(x * x for x in a)), math.sqrt(sum(y * y for y in b))
return 1 - dot / (na * nb) if na and nb else 1.0
scored = [(cid, cos_dist(v, vector)) for cid, (bid, v, _) in self.vectors.items()
if bid in board_ids]
return sorted(scored, key=lambda x: x[1])[:limit]
# live -----------------------------------------------------------------------------------
def latest_event(self, board_id) -> Optional[Event]:
return next((e for e in reversed(self.events) if e.board_id == board_id), None)
# comments & notifications ---------------------------------------------------------------
def add_comment(self, comment, event) -> None:
self.comment_rows.append(dict(comment))
self.events.append(copy.deepcopy(event))
def comments(self, card_id) -> list:
return [dict(c) for c in self.comment_rows if c["card_id"] == card_id]
def add_notifications(self, items) -> None:
self.notes.extend(dict(n) for n in items)
def notifications_for(self, email, limit=50) -> list:
return [dict(n) for n in reversed(self.notes) if n["email"] == email][:limit]
def mark_read(self, email, ids, at) -> int:
n = 0
for note in self.notes:
if note["email"] == email and not note.get("read_at") and (ids is None or note["id"] in ids):
note["read_at"] = at
n += 1
return n
# attachments ----------------------------------------------------------------------------
def add_attachment(self, att, event) -> None:
self.atts[att["id"]] = dict(att)
self.events.append(copy.deepcopy(event))
def attachments(self, card_id) -> list:
return sorted((dict(a) for a in self.atts.values() if a["card_id"] == card_id),
key=lambda a: a["created_at"])
def get_attachment(self, att_id) -> Optional[dict]:
a = self.atts.get(att_id)
return dict(a) if a else None
def delete_attachment(self, att_id, event) -> None:
self.atts.pop(att_id, None)
self.events.append(copy.deepcopy(event))
# archive --------------------------------------------------------------------------------
def archive_card(self, card_id, board_id, by, at, event) -> None:
self.archived[card_id] = {"card_id": card_id, "board_id": board_id, "by": by, "at": at}
self.events.append(copy.deepcopy(event))
def restore_card(self, card_id, event) -> None:
self.archived.pop(card_id, None)
self.events.append(copy.deepcopy(event))
def archived_on(self, board_id) -> dict:
return {k: dict(v) for k, v in self.archived.items() if v["board_id"] == board_id}
def card_counts(self, board_id) -> dict:
out = {}
for c in self.comment_rows:
if c["board_id"] == board_id:
out.setdefault(c["card_id"], {"comments": 0, "files": 0})["comments"] += 1
for a in self.atts.values():
if a["board_id"] == board_id:
out.setdefault(a["card_id"], {"comments": 0, "files": 0})["files"] += 1
return out
class MemoryObjects:
"""Attachment bytes in memory, with the same contract as DataK3 object storage."""
def __init__(self):
self.blobs = {}
def put(self, key, data, content_type) -> None:
self.blobs[key] = (bytes(data), content_type)
def get(self, key) -> bytes:
return self.blobs[key][0]
def delete(self, key) -> None:
self.blobs.pop(key, None)