theaio/theaio-boards
Laneway, team task boards you can run from Claude or ChatGPT. FastAPI + DataK3 + appid + Models + MCP.
A remix template · 0 remixes · Remix theaio/theaio-boards
app/boards/infrastructure/repository.py
"""Boards, lists, cards, history, links and embeddings stored in DataK3 tables (schema.sql).
Writes are keyed upserts, so a retry is safe: INSERT ... ON CONFLICT (id) DO UPDATE SET ... .
Two writes are deliberately plain INSERTs:
- insert_card CLAIMS a card number: its id is board_id + ':' + number, so when two people add
a card at the same moment, one INSERT wins and the other gets 23505 and tries the next number
(an upsert would let both "win" and one card would overwrite the other);
- record() appends to card_history, which is never updated.
Insert-only columns (created_by, created_at) are left out of upserts on purpose.
"""
import json
import uuid
from typing import Optional
import psycopg
from boards.domain.model import DEFAULT_LISTS, Board, BoardList, Card, HistoryEntry, Role
from boards.infrastructure.dodil import connect
CARD_COLUMNS = ("id, board_id, num, list_id, title, description, assignee, due_date, labels, "
"pos, created_by")
def _card(row) -> Card:
(cid, board_id, num, list_id, title, description, assignee, due_date, labels, pos,
created_by) = row
return Card(id=cid, board_id=board_id, number=int(num or 0), list_id=list_id or "",
title=title, description=description or "", assignee=assignee,
due_date=due_date, labels=json.loads(labels) if labels else [],
pos=int(pos or 0), created_by=created_by or "")
def _vector_text(vector) -> str:
"""pgvector's text form, bound as an untyped parameter so DataK3 reads it as VECTOR."""
return "[" + ",".join(f"{float(x):.7g}" for x in vector) + "]"
class SqlBoards:
# --- boards, members, lists -------------------------------------------------------------
def _load(self, c, board_id: str) -> Optional[Board]:
row = c.execute("SELECT id, title, owner_email FROM boards WHERE id = %s",
(board_id,)).fetchone()
if not row:
return None
members = c.execute("SELECT email, role FROM memberships WHERE board_id = %s",
(board_id,)).fetchall()
lists = c.execute("SELECT id, board_id, name, pos FROM lists WHERE board_id = %s",
(board_id,)).fetchall()
if not lists:
lists = self._adopt_starter_board(c, board_id)
return Board(id=row[0], title=row[1], owner_email=row[2],
members={e: Role(r) for e, r in members if r != Role.OWNER.value},
lists=[BoardList(i, b, n, int(p)) for i, b, n, p in lists])
def _adopt_starter_board(self, c, board_id: str) -> list:
"""A board made before the upgrade (migrations/001) has no list rows: its cards name one
of the starter's fixed lists. Give it real lists and number its cards, once."""
lists = [(f"{board_id}:{pos}", board_id, name, pos)
for pos, name in enumerate(DEFAULT_LISTS)]
for row in lists:
c.execute("INSERT INTO lists (id, board_id, name, pos) VALUES (%s, %s, %s, %s) "
"ON CONFLICT (id) DO UPDATE SET board_id = EXCLUDED.board_id", row)
by_name = {name: lid for lid, _, name, _ in lists}
old = c.execute("SELECT id, list_name FROM cards WHERE board_id = %s AND list_id IS NULL "
"ORDER BY updated_at", (board_id,)).fetchall()
top = c.execute("SELECT max(num) AS n FROM cards WHERE board_id = %s",
(board_id,)).fetchone()[0] or 0
for i, (cid, list_name) in enumerate(old, start=1):
c.execute("UPDATE cards SET list_id = %s, num = %s, pos = %s, labels = '[]' "
"WHERE id = %s",
(by_name.get(list_name, lists[0][0]), top + i, top + i, cid))
return lists
def get(self, board_id: str) -> Optional[Board]:
with connect() as c:
return self._load(c, board_id)
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 "
"UNION SELECT board_id FROM memberships WHERE email = %s", (email, email))]
return [b for b in (self._load(c, i) for i in ids) if b]
def save_board(self, board: Board) -> None:
with connect() as c:
c.execute("INSERT INTO boards (id, title, owner_email, created_at) "
"VALUES (%s, %s, %s, now()) ON CONFLICT (id) DO UPDATE SET "
"title = EXCLUDED.title, owner_email = EXCLUDED.owner_email",
(board.id, board.title, board.owner_email))
for email, role in board.members.items():
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.value))
for lst in board.lists:
c.execute("INSERT INTO lists (id, board_id, name, pos) VALUES (%s, %s, %s, %s) "
"ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, "
"pos = EXCLUDED.pos", (lst.id, board.id, lst.name, lst.pos))
def remove_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 log(self, board_id: str, card_id: Optional[str], actor: str, action: str) -> None:
with connect() as c:
c.execute("INSERT INTO activity (id, board_id, card_id, actor, action, event_at) "
"VALUES (%s, %s, %s, %s, %s, now())",
(uuid.uuid4().hex, board_id, card_id, actor, action))
# --- cards ------------------------------------------------------------------------------
def get_card(self, card_id: str) -> Optional[Card]:
with connect() as c:
r = c.execute(f"SELECT {CARD_COLUMNS} 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]:
with connect() as c:
r = c.execute(f"SELECT {CARD_COLUMNS} FROM cards WHERE board_id = %s AND num = %s",
(board_id, number)).fetchone()
return _card(r) if r else None
def cards_on(self, board_id: str) -> list:
with connect() as c:
rows = c.execute(f"SELECT {CARD_COLUMNS} FROM cards WHERE board_id = %s "
"ORDER BY pos, num", (board_id,)).fetchall()
return [_card(r) for r in rows]
def max_card_number(self, board_id: str) -> int:
with connect() as c:
r = c.execute("SELECT max(num) AS n FROM cards WHERE board_id = %s",
(board_id,)).fetchone()
return int(r[0] or 0) if r else 0
def insert_card(self, card: Card) -> bool:
try:
with connect() as c:
c.execute("INSERT INTO cards (id, board_id, num, list_id, list_name, title, "
"description, assignee, due_date, labels, pos, created_by, created_at, "
"updated_at) VALUES (%s, %s, %s, %s, '', %s, %s, %s, %s, %s, %s, %s, "
"now(), now())",
(card.id, card.board_id, card.number, card.list_id, card.title,
card.description, card.assignee, card.due_date,
json.dumps(card.labels), card.pos, card.created_by))
return True
except psycopg.errors.UniqueViolation: # 23505: that number was just taken
return False
def save_card(self, card: Card) -> None:
with connect() as c:
c.execute("INSERT INTO cards (id, board_id, num, list_id, list_name, title, "
"description, assignee, due_date, labels, pos, updated_at) "
"VALUES (%s, %s, %s, %s, '', %s, %s, %s, %s, %s, %s, now()) "
"ON CONFLICT (id) DO UPDATE SET list_id = EXCLUDED.list_id, "
"title = EXCLUDED.title, description = EXCLUDED.description, "
"assignee = EXCLUDED.assignee, due_date = EXCLUDED.due_date, "
"labels = EXCLUDED.labels, pos = EXCLUDED.pos, "
"updated_at = EXCLUDED.updated_at",
(card.id, card.board_id, card.number, card.list_id, card.title,
card.description, card.assignee, card.due_date,
json.dumps(card.labels), card.pos))
# --- history (append-only) --------------------------------------------------------------
def record(self, entries: list) -> None:
if not entries:
return
with connect() as c:
for e in entries:
c.execute("INSERT INTO card_history (id, board_id, card_id, actor, changed_at, "
"field_name, old_value, new_value) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s)",
(e.id, e.board_id, e.card_id, e.actor, e.at, e.field, e.old, e.new))
def _history(self, where: str, args: tuple) -> list:
with connect() as c:
rows = c.execute("SELECT id, board_id, card_id, actor, changed_at, field_name, "
f"old_value, new_value FROM card_history WHERE {where} "
"ORDER BY changed_at", args).fetchall()
return [HistoryEntry(*r) for r in rows]
def history(self, card_id: str) -> list:
return self._history("card_id = %s", (card_id,))
def board_history(self, board_id: str, since) -> list:
return self._history("board_id = %s AND changed_at >= %s", (board_id, since))
# --- "blocked by" links -----------------------------------------------------------------
def links_on(self, board_id: str) -> list:
with connect() as c:
rows = c.execute("SELECT card_id, blocked_by FROM card_links WHERE board_id = %s",
(board_id,)).fetchall()
return [(a, b) for a, b in rows]
def add_link(self, board_id: str, card_id: str, blocked_by: str) -> None:
with connect() as c:
c.execute("INSERT INTO card_links (id, board_id, card_id, blocked_by) "
"VALUES (%s, %s, %s, %s) ON CONFLICT (id) DO UPDATE SET "
"board_id = EXCLUDED.board_id",
(f"{card_id}<{blocked_by}", board_id, card_id, blocked_by))
def remove_link(self, card_id: str, blocked_by: str) -> None:
with connect() as c:
c.execute("DELETE FROM card_links WHERE id = %s", (f"{card_id}<{blocked_by}",))
# --- embeddings (pgvector on DataK3) ----------------------------------------------------
def embedding_digests(self, board_ids: list) -> dict:
if not board_ids:
return {}
with connect() as c:
rows = c.execute("SELECT card_id, digest FROM card_embeddings "
"WHERE board_id = ANY(%s)", (list(board_ids),)).fetchall()
return dict(rows)
def save_embedding(self, card: Card, digest: str, vector: list) -> None:
with connect() as c:
c.execute("INSERT INTO card_embeddings (card_id, board_id, digest, embedding) "
"VALUES (%s, %s, %s, %s::vector) ON CONFLICT (card_id) DO UPDATE SET "
"digest = EXCLUDED.digest, embedding = EXCLUDED.embedding",
(card.id, card.board_id, digest, _vector_text(vector)))
def vector_of(self, card_id: str) -> Optional[list]:
with connect() as c:
r = c.execute("SELECT embedding FROM card_embeddings WHERE card_id = %s",
(card_id,)).fetchone()
if not r or r[0] is None:
return None
return json.loads(r[0]) if isinstance(r[0], str) else [float(x) for x in r[0]]
def nearest(self, board_ids: list, vector: list, exclude_id: str, limit: int) -> list:
"""KNN by cosine distance (<=>), only among the given boards: visibility holds here too.
The text parameters are cast to vector / text[]: pgvector has no text <=> vector operator, and an empty
list has no element type of its own."""
if not board_ids:
return []
v = _vector_text(vector)
with connect() as c:
rows = c.execute("SELECT card_id, embedding <=> %s::vector AS dist FROM card_embeddings "
"WHERE board_id = ANY(%s::text[]) AND card_id <> %s "
"ORDER BY embedding <=> %s::vector LIMIT %s",
(v, list(board_ids), exclude_id, v, int(limit))).fetchall()
return [(cid, float(d)) for cid, d in rows]