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/repository.py
"""Boards and cards in DataK3 tables (schema.sql), over the Postgres wire.
Three DataK3 facts this file is built around:
- A card number is CLAIMED by a plain INSERT: two people adding a card at the same moment can't
both get #12; the loser gets a unique violation and the use case tries #13.
- A card write is a conditional UPDATE (... WHERE version = the one you read): a stale write
matches no row, so nothing is overwritten. No upsert for cards, ever.
- The card row and its history line are written in one transaction.
"""
import json
from typing import Optional
import psycopg
from boards.domain.model import Board, BoardList, Card, Event, Role
from boards.infrastructure.dodil import connect
CARD_COLS = ("id, board_id, number, list_id, title, description, assignee, due_date, labels, "
"position, version, created_by, created_at, updated_at")
def _card(r) -> Card:
return Card(id=r[0], board_id=r[1], number=int(r[2]), list_id=r[3], title=r[4],
description=r[5] or "", assignee=r[6], due_date=r[7],
labels=json.loads(r[8]) if r[8] else [], position=float(r[9] or 0),
version=int(r[10] or 1), created_by=r[11] or "", created_at=r[12],
updated_at=r[13])
def _event(r) -> Event:
return Event(id=r[0], board_id=r[1], card_id=r[2], actor=r[3], action=r[4],
summary=r[5] or "", changes=json.loads(r[6]) if r[6] else {}, at=r[7])
def _insert_event(c, e: Event) -> None:
c.execute("INSERT INTO card_events (id, board_id, card_id, actor, action, summary, changes, "
"event_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)",
(e.id, e.board_id, e.card_id, e.actor, e.action, e.summary,
json.dumps(e.changes, default=str), e.at))
class SqlBoards:
# boards ---------------------------------------------------------------------------------
def _load(self, c, rows) -> list:
if not rows:
return []
ids = [r[0] for r in rows]
members = c.execute("SELECT board_id, email, role FROM memberships "
"WHERE board_id = ANY(%s)", (ids,)).fetchall()
lists = c.execute("SELECT id, board_id, name, position, is_done FROM board_lists "
"WHERE board_id = ANY(%s)", (ids,)).fetchall()
out = []
for r in rows:
b = Board(id=r[0], title=r[1], owner_email=r[2], created_at=r[3])
b.members = {m[1]: Role(m[2]) for m in members if m[0] == b.id}
b.lists = sorted((BoardList(id=l[0], board_id=l[1], name=l[2],
position=float(l[3] or 0), is_done=bool(l[4]))
for l in lists if l[1] == b.id), key=lambda l: l.position)
out.append(b)
return out
def get(self, board_id: str) -> Optional[Board]:
with connect() as c:
rows = c.execute("SELECT id, title, owner_email, created_at FROM boards "
"WHERE id = %s", (board_id,)).fetchall()
found = self._load(c, rows)
return found[0] if found else None
def boards_for(self, email: str) -> list:
with connect() as c:
ids = {r[0] for r in c.execute("SELECT id FROM boards WHERE owner_email = %s",
(email,))}
ids |= {r[0] for r in c.execute("SELECT board_id FROM memberships WHERE email = %s",
(email,))}
if not ids:
return []
rows = c.execute("SELECT id, title, owner_email, created_at FROM boards "
"WHERE id = ANY(%s)", (list(ids),)).fetchall()
return self._load(c, rows)
def create_board(self, board: Board) -> None:
with connect() as c, c.transaction():
c.execute("INSERT INTO boards (id, title, owner_email, created_at) "
"VALUES (%s, %s, %s, %s)",
(board.id, board.title, board.owner_email, board.created_at))
for l in board.lists:
c.execute("INSERT INTO board_lists (id, board_id, name, position, is_done) "
"VALUES (%s, %s, %s, %s, %s)",
(l.id, l.board_id, l.name, l.position, l.is_done))
def rename_board(self, board_id: str, title: str) -> None:
with connect() as c:
c.execute("UPDATE boards SET title = %s WHERE id = %s", (title, board_id))
def save_member(self, board_id: str, email: str, role: str) -> None:
with connect() as c:
c.execute("INSERT INTO memberships (id, board_id, email, role) VALUES (%s, %s, %s, %s) "
"ON CONFLICT (id) DO UPDATE SET role = EXCLUDED.role",
(f"{board_id}:{email}", board_id, email, role))
def delete_member(self, board_id: str, email: str) -> None:
with connect() as c:
c.execute("DELETE FROM memberships WHERE id = %s", (f"{board_id}:{email}",))
def save_list(self, lst: BoardList) -> None:
with connect() as c:
c.execute("INSERT INTO board_lists (id, board_id, name, position, is_done) "
"VALUES (%s, %s, %s, %s, %s) ON CONFLICT (id) DO UPDATE SET "
"name = EXCLUDED.name, position = EXCLUDED.position, "
"is_done = EXCLUDED.is_done",
(lst.id, lst.board_id, lst.name, lst.position, lst.is_done))
# cards ----------------------------------------------------------------------------------
def get_card(self, card_id: str) -> Optional[Card]:
with connect() as c:
r = c.execute(f"SELECT {CARD_COLS} FROM cards WHERE id = %s", (card_id,)).fetchone()
return _card(r) if r 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:
with connect() as c:
rows = c.execute(f"SELECT {CARD_COLS} FROM cards WHERE board_id = %s",
(board_id,)).fetchall()
return sorted((_card(r) for r in rows), key=lambda x: (x.position, x.number))
def cards_assigned_to(self, email: str) -> list:
with connect() as c:
rows = c.execute(f"SELECT {CARD_COLS} FROM cards WHERE assignee = %s",
(email,)).fetchall()
return [_card(r) for r in rows]
def next_number(self, board_id: str) -> int:
with connect() as c:
r = c.execute("SELECT max(number) FROM cards WHERE board_id = %s",
(board_id,)).fetchone()
return int(r[0] or 0) + 1
def insert_card(self, card: Card, event: Event) -> bool:
try:
with connect() as c, c.transaction():
c.execute(f"INSERT INTO cards ({CARD_COLS}) VALUES "
"(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
(card.id, card.board_id, card.number, card.list_id, card.title,
card.description, card.assignee, card.due_date,
json.dumps(card.labels), card.position, card.version,
card.created_by, card.created_at, card.updated_at))
_insert_event(c, event)
return True
except psycopg.errors.UniqueViolation:
return False
except psycopg.Error as e: # some adapters report the clash without the SQLSTATE class
if "duplicate key" in str(e).lower() or "already exists" in str(e).lower():
return False
raise
def update_card(self, card: Card, expected_version: int, event: Event) -> bool:
try:
self._update_or_raise(card, expected_version, event)
return True
except _Stale:
return False
def _update_or_raise(self, card: Card, expected_version: int, event: Event) -> None:
"""Raising inside the transaction rolls back the history line with the card."""
with connect() as c, c.transaction():
cur = c.execute("UPDATE cards SET list_id = %s, title = %s, description = %s, "
"assignee = %s, due_date = %s, labels = %s, position = %s, "
"version = %s, updated_at = %s WHERE id = %s AND version = %s",
(card.list_id, card.title, card.description, card.assignee,
card.due_date, json.dumps(card.labels), card.position, card.version,
card.updated_at, card.id, expected_version))
if cur.rowcount != 1:
raise _Stale()
_insert_event(c, event)
def unassign_all(self, board_id: str, email: str, event_for) -> int:
n = 0
for card in [c for c in self.cards_on(board_id) if c.assignee == email]:
old = card.version
card.assignee, card.version = None, card.version + 1
# if someone changed it meanwhile, their write already decided the assignee
n += self.update_card(card, old, event_for(card))
return n
# history --------------------------------------------------------------------------------
def log(self, event: Event) -> None:
with connect() as c:
_insert_event(c, event)
def history(self, card_id: str, limit: int = 50) -> list:
with connect() as c:
rows = c.execute("SELECT id, board_id, card_id, actor, action, summary, changes, "
"event_at FROM card_events WHERE card_id = %s "
"ORDER BY event_at DESC LIMIT %s", (card_id, limit)).fetchall()
return [_event(r) for r in rows]
def board_activity(self, board_id: str, limit: int = 50) -> list:
with connect() as c:
rows = c.execute("SELECT id, board_id, card_id, actor, action, summary, changes, "
"event_at FROM card_events WHERE board_id = %s "
"ORDER BY event_at DESC LIMIT %s", (board_id, limit)).fetchall()
return [_event(r) for r in rows]
# dependencies ---------------------------------------------------------------------------
def blockers_on(self, board_id: str) -> dict:
with connect() as c:
rows = c.execute("SELECT card_id, blocker_id FROM card_dependencies "
"WHERE board_id = %s", (board_id,)).fetchall()
out = {}
for card_id, blocker_id in rows:
out.setdefault(card_id, set()).add(blocker_id)
return out
def add_dependency(self, board_id: str, card_id: str, blocker_id: str) -> None:
with connect() as c:
c.execute("INSERT INTO card_dependencies (id, board_id, card_id, blocker_id) "
"VALUES (%s, %s, %s, %s) ON CONFLICT (id) DO NOTHING",
(f"{card_id}<{blocker_id}", board_id, card_id, blocker_id))
def remove_dependency(self, card_id: str, blocker_id: str) -> None:
with connect() as c:
c.execute("DELETE FROM card_dependencies WHERE id = %s", (f"{card_id}<{blocker_id}",))
# checklists -----------------------------------------------------------------------------
def checklist(self, card_id: str) -> list:
with connect() as c:
rows = c.execute("SELECT id, text, done, position FROM checklist_items "
"WHERE card_id = %s", (card_id,)).fetchall()
return sorted(({"id": r[0], "text": r[1], "done": bool(r[2]),
"position": float(r[3] or 0)} for r in rows),
key=lambda i: i["position"])
def save_checklist_item(self, card_id, item_id, text, done, position) -> None:
with connect() as c:
c.execute("INSERT INTO checklist_items (id, card_id, text, done, position) "
"VALUES (%s, %s, %s, %s, %s) ON CONFLICT (id) DO UPDATE SET "
"text = EXCLUDED.text, done = EXCLUDED.done, position = EXCLUDED.position",
(item_id, card_id, text, bool(done), float(position)))
def delete_checklist_item(self, card_id, item_id) -> None:
with connect() as c:
c.execute("DELETE FROM checklist_items WHERE id = %s AND card_id = %s",
(item_id, card_id))
# meaning search -------------------------------------------------------------------------
def save_embedding(self, card_id, board_id, vector, text_hash) -> None:
literal = "[" + ",".join(f"{float(x):.6f}" for x in vector) + "]"
with connect() as c:
c.execute("INSERT INTO card_vectors (card_id, board_id, text_hash, embedding) "
"VALUES (%s, %s, %s, %s) ON CONFLICT (card_id) DO UPDATE SET "
"board_id = EXCLUDED.board_id, text_hash = EXCLUDED.text_hash, "
"embedding = EXCLUDED.embedding", (card_id, board_id, text_hash, literal))
def embedding_hash(self, card_id) -> Optional[str]:
with connect() as c:
r = c.execute("SELECT text_hash FROM card_vectors WHERE card_id = %s",
(card_id,)).fetchone()
return r[0] if r else None
def similar_cards(self, board_ids, vector, limit) -> list:
if not board_ids:
return []
literal = "[" + ",".join(f"{float(x):.6f}" for x in vector) + "]"
with connect() as c:
rows = c.execute("SELECT card_id, embedding <=> %s AS d FROM card_vectors "
"WHERE board_id = ANY(%s) ORDER BY embedding <=> %s LIMIT %s",
(literal, list(board_ids), literal, int(limit))).fetchall()
return [(r[0], float(r[1])) for r in rows]
# live -----------------------------------------------------------------------------------
def latest_event(self, board_id: str) -> Optional[Event]:
with connect() as c:
r = c.execute("SELECT id, board_id, card_id, actor, action, summary, changes, "
"event_at FROM card_events WHERE board_id = %s "
"ORDER BY event_at DESC LIMIT 1", (board_id,)).fetchone()
return _event(r) if r else None
# comments & notifications ---------------------------------------------------------------
def add_comment(self, comment: dict, event: Event) -> None:
with connect() as c, c.transaction():
c.execute("INSERT INTO card_comments (id, board_id, card_id, author, body, mentions, "
"created_at) VALUES (%s, %s, %s, %s, %s, %s, %s)",
(comment["id"], comment["board_id"], comment["card_id"], comment["author"],
comment["body"], json.dumps(comment["mentions"]), comment["created_at"]))
_insert_event(c, event)
def comments(self, card_id: str) -> list:
with connect() as c:
rows = c.execute("SELECT id, board_id, card_id, author, body, mentions, created_at "
"FROM card_comments WHERE card_id = %s ORDER BY created_at",
(card_id,)).fetchall()
return [{"id": r[0], "board_id": r[1], "card_id": r[2], "author": r[3], "body": r[4],
"mentions": json.loads(r[5]) if r[5] else [], "created_at": r[6]} for r in rows]
def add_notifications(self, items: list) -> None:
if not items:
return
with connect() as c, c.transaction():
for n in items:
c.execute("INSERT INTO notifications (id, email, board_id, card_id, kind, actor, "
"summary, created_at, read_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, "
"NULL)", (n["id"], n["email"], n["board_id"], n["card_id"], n["kind"],
n["actor"], n["summary"], n["created_at"]))
def notifications_for(self, email: str, limit: int = 50) -> list:
with connect() as c:
rows = c.execute("SELECT id, email, board_id, card_id, kind, actor, summary, "
"created_at, read_at FROM notifications WHERE email = %s "
"ORDER BY created_at DESC LIMIT %s", (email, limit)).fetchall()
keys = ("id", "email", "board_id", "card_id", "kind", "actor", "summary", "created_at",
"read_at")
return [dict(zip(keys, r)) for r in rows]
def mark_read(self, email: str, ids, at) -> int:
with connect() as c:
if ids is None:
cur = c.execute("UPDATE notifications SET read_at = %s WHERE email = %s "
"AND read_at IS NULL", (at, email))
else:
cur = c.execute("UPDATE notifications SET read_at = %s WHERE email = %s "
"AND read_at IS NULL AND id = ANY(%s)", (at, email, list(ids)))
return int(cur.rowcount or 0)
# attachments ----------------------------------------------------------------------------
_ATT = ("id", "board_id", "card_id", "name", "content_type", "size", "object_key",
"uploaded_by", "created_at")
def add_attachment(self, att: dict, event: Event) -> None:
with connect() as c, c.transaction():
c.execute(f"INSERT INTO card_attachments ({', '.join(self._ATT)}) VALUES "
"(%s, %s, %s, %s, %s, %s, %s, %s, %s)", tuple(att[k] for k in self._ATT))
_insert_event(c, event)
def attachments(self, card_id: str) -> list:
with connect() as c:
rows = c.execute(f"SELECT {', '.join(self._ATT)} FROM card_attachments "
"WHERE card_id = %s ORDER BY created_at", (card_id,)).fetchall()
return [dict(zip(self._ATT, r)) for r in rows]
def get_attachment(self, att_id: str) -> Optional[dict]:
with connect() as c:
r = c.execute(f"SELECT {', '.join(self._ATT)} FROM card_attachments WHERE id = %s",
(att_id,)).fetchone()
return dict(zip(self._ATT, r)) if r else None
def delete_attachment(self, att_id: str, event: Event) -> None:
with connect() as c, c.transaction():
c.execute("DELETE FROM card_attachments WHERE id = %s", (att_id,))
_insert_event(c, event)
# archive --------------------------------------------------------------------------------
def archive_card(self, card_id: str, board_id: str, by: str, at, event: Event) -> None:
with connect() as c, c.transaction():
c.execute("INSERT INTO archived_cards (card_id, board_id, archived_by, archived_at) "
"VALUES (%s, %s, %s, %s) ON CONFLICT (card_id) DO NOTHING",
(card_id, board_id, by, at))
_insert_event(c, event)
def restore_card(self, card_id: str, event: Event) -> None:
with connect() as c, c.transaction():
c.execute("DELETE FROM archived_cards WHERE card_id = %s", (card_id,))
_insert_event(c, event)
def archived_on(self, board_id: str) -> dict:
with connect() as c:
rows = c.execute("SELECT card_id, board_id, archived_by, archived_at FROM "
"archived_cards WHERE board_id = %s", (board_id,)).fetchall()
return {r[0]: {"card_id": r[0], "board_id": r[1], "by": r[2], "at": r[3]} for r in rows}
def card_counts(self, board_id: str) -> dict:
out = {}
with connect() as c:
for table, key in (("card_comments", "comments"), ("card_attachments", "files")):
for card_id, n in c.execute(f"SELECT card_id, count(*) FROM {table} "
"WHERE board_id = %s GROUP BY card_id", (board_id,)):
out.setdefault(card_id, {"comments": 0, "files": 0})[key] = int(n)
return out
def counts(self) -> dict:
tables = ("boards", "memberships", "board_lists", "cards", "card_events",
"card_dependencies", "checklist_items", "card_vectors", "card_comments",
"notifications", "card_attachments", "archived_cards")
with connect() as c:
return {t: int(c.execute(f"SELECT count(*) FROM {t}").fetchone()[0]) for t in tables}
class _Stale(Exception):
"""The conditional UPDATE matched no row: the card changed since it was read."""