sirenamc/stackline
Stackline, a team task board you run from the web or from Claude/ChatGPT. FastAPI + DataK3 + appid + Models + MCP.
A remix template · 0 remixes · Remix sirenamc/stackline
app/boards/infrastructure/repository.py
"""Boards and cards stored in DataK3 tables (schema.sql). Writes are keyed upserts, so a retry is
safe: INSERT ... ON CONFLICT (id) DO UPDATE SET col = EXCLUDED.col."""
import json
import uuid
from datetime import date, datetime
from typing import Optional
import psycopg
from boards.application.ports import Event
from boards.domain.model import Board, BoardList, Card, Role
from boards.infrastructure.dodil import db
TABLES = ("boards", "memberships", "lists", "cards", "card_numbers", "activity", "embeddings")
CARD_COLS = ("number, board_id, list_id, title, description, assignee, due_date, labels, "
"blocked_by, checklist, position")
def _date(v) -> Optional[date]:
"""DataK3 (DuckDB underneath) may hand a DATE back as a timestamp."""
return v.date() if isinstance(v, datetime) else v
def _card(r) -> Card:
return Card(number=r[0], board_id=r[1], list_id=r[2], title=r[3], description=r[4] or "",
assignee=r[5], due_date=_date(r[6]), labels=tuple(json.loads(r[7] or "[]")),
blocked_by=tuple(json.loads(r[8] or "[]")),
checklist=tuple((t, bool(d)) for t, d in json.loads(r[9] or "[]")),
position=float(r[10] or 0))
def _card_row(c: Card) -> tuple:
return (c.number, c.board_id, c.list_id, c.title, c.description, c.assignee, c.due_date,
json.dumps(list(c.labels)), json.dumps(list(c.blocked_by)),
json.dumps([[t, d] for t, d in c.checklist]), c.position)
class SqlBoards:
def get(self, board_id: str) -> Optional[Board]:
with db() as c:
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, name, position, is_done FROM lists "
"WHERE board_id = %s ORDER BY position", (board_id,)).fetchall()
return Board(id=row[0], title=row[1], owner_email=row[2],
members={e: Role(r) for e, r in members},
lists=[BoardList(i, n, p, bool(d)) for i, n, p, d in lists])
def boards_for(self, email: str) -> list:
with db() 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))]
boards = [b for b in (self.get(i) for i in ids) if b]
return sorted(boards, key=lambda b: b.title.lower())
def save_board(self, board: Board) -> None:
with db() 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))
def save_list(self, board_id: str, lst: BoardList) -> None:
with db() as c:
c.execute("INSERT INTO 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, board_id, lst.name, lst.position, lst.is_done))
def save_membership(self, board_id: str, email: str, role: Role) -> None:
with db() 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.value))
def remove_membership(self, board_id: str, email: str) -> None:
with db() as c:
c.execute("DELETE FROM memberships WHERE id = %s", (f"{board_id}:{email}",))
def get_card(self, number: int) -> Optional[Card]:
with db() as c:
r = c.execute(f"SELECT {CARD_COLS} FROM cards WHERE number = %s",
(number,)).fetchone()
return _card(r) if r else None
def cards_on(self, board_id: str) -> list:
with db() as c:
rows = c.execute(f"SELECT {CARD_COLS} FROM cards WHERE board_id = %s",
(board_id,)).fetchall()
return [_card(r) for r in rows]
def insert_card(self, card: Card) -> Card:
"""No sequences in DataK3: claim the next number in card_numbers with a PLAIN insert on
its primary key (an upsert would let every concurrent writer win). If someone else
claimed it first, we get a unique violation and try the next one.
Claimed numbers stay claimed after a card is deleted, so a new card never inherits an
old card's history, links or embedding."""
with db() as c:
# After the first claim, the next number is almost always this copy's last + 1, which
# saves four MAX() queries a card. The first claim looks past every table.
last = getattr(self, "_last", 0)
n = last + 1 if last else self._next_number(c)
for _ in range(20):
try:
c.execute("INSERT INTO card_numbers (number, board_id, claimed_at) "
"VALUES (%s, %s, now())", (n, card.board_id))
except psycopg.errors.UniqueViolation:
n = self._next_number(c)
continue
card.number = self._last = n
c.execute(f"INSERT INTO cards ({CARD_COLS}, updated_at) VALUES "
"(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, now())", _card_row(card))
return card
raise RuntimeError("could not claim a card number; try again")
@staticmethod
def _next_number(c) -> int:
"""Past every number any table has seen, not only card_numbers: a bucket filled before
card_numbers existed (or edited by hand) must not hand out a used number again."""
highest = [c.execute(q).fetchone()[0] or 0 for q in (
"SELECT MAX(number) FROM card_numbers", "SELECT MAX(number) FROM cards",
"SELECT MAX(card_number) FROM activity", "SELECT MAX(card_number) FROM embeddings")]
return max(highest) + 1
def save_card(self, card: Card) -> None:
with db() as c:
c.execute(f"INSERT INTO cards ({CARD_COLS}, updated_at) VALUES "
"(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, now()) "
"ON CONFLICT (number) 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, blocked_by = EXCLUDED.blocked_by, "
"checklist = EXCLUDED.checklist, position = EXCLUDED.position, "
"updated_at = EXCLUDED.updated_at", _card_row(card))
def delete_card(self, number: int) -> None:
with db() as c:
c.execute("DELETE FROM cards WHERE number = %s", (number,))
c.execute("DELETE FROM embeddings WHERE card_number = %s", (number,))
def delete_board(self, board_id: str) -> None:
"""Everything on a board goes, except its claimed card numbers: those are never handed
out again."""
with db() as c:
for table in ("embeddings", "activity", "cards", "lists", "memberships"):
c.execute(f"DELETE FROM {table} WHERE board_id = %s", (board_id,))
c.execute("DELETE FROM boards WHERE id = %s", (board_id,))
def log(self, e: Event) -> None:
with db() as c:
c.execute("INSERT INTO activity (id, board_id, card_number, actor, action, field, "
"before_value, after_value, event_at) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, now())",
(uuid.uuid4().hex, e.board_id, e.card_number, e.actor, e.action, e.field,
e.before, e.after))
def _events(self, where: str, args: tuple, limit: int) -> list:
with db() as c:
rows = c.execute("SELECT board_id, card_number, actor, action, field, before_value, "
f"after_value, event_at FROM activity WHERE {where} "
"ORDER BY event_at DESC LIMIT %s", (*args, limit)).fetchall()
return [Event(*r) for r in rows]
def history(self, board_id: str, card_number: int) -> list:
return self._events("board_id = %s AND card_number = %s", (board_id, card_number), 200)
def activity(self, board_id: str, limit: int) -> list:
return self._events("board_id = %s", (board_id,), limit)
def save_embedding(self, number: int, text_hash: str, vector: list) -> None:
card = self.get_card(number)
if card is None:
return
with db() as c:
c.execute("INSERT INTO embeddings (card_number, board_id, text_hash, vector) "
"VALUES (%s, %s, %s, %s) ON CONFLICT (card_number) DO UPDATE SET "
"board_id = EXCLUDED.board_id, text_hash = EXCLUDED.text_hash, "
"vector = EXCLUDED.vector",
(number, card.board_id, text_hash, json.dumps(vector)))
def counts(self) -> dict:
"""Rows per table, read with the app's own credential: the deploy self-check."""
with db() as c:
return {t: c.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0] for t in TABLES}
def embeddings(self, board_ids: list) -> dict:
if not board_ids:
return {}
with db() as c:
rows = c.execute("SELECT card_number, text_hash, vector FROM embeddings "
"WHERE board_id = ANY(%s)", (list(board_ids),)).fetchall()
return {n: (h, json.loads(v)) for n, h, v in rows}