sawera/sawera-boards
Sawera Boards - a team task board for the Dodil Boards Challenge. Boards, lists, cards, roles, activity log, MCP actions and AI extras. FastAPI + DataK3 + appid + Models + MCP.
A remix template · 0 remixes · Remix sawera/sawera-boards
app/boards/infrastructure/repository.py
"""Boards, lists 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.
Lists and members are saved as full sets for their board: the app deletes rows that vanished
and upserts the rest, all inside the same connection, so a rename or removal can't half-apply.
"""
import json
import sys
import uuid
from typing import Optional
from boards.domain.model import Board, BoardList, Card, Comment, Role
from boards.infrastructure.dodil import connect
_CARD_COLS = ("id, board_id, list_id, title, description, assignee, due_date, "
"labels, position, blocked_by")
# A remix creates the bucket's tables from the template's schema.sql at remix time, so a copy
# made from an older template keeps the old table shape after the code moves on. These
# statements upgrade any older bucket to the shape this code expects. Each one is idempotent
# (IF NOT EXISTS, or a constraint drop that only applies where the old column exists) and runs
# best-effort on its own connection: on a fresh bucket from the current schema.sql they are
# no-ops, and a statement that does not apply must never stop the app from starting.
_MIGRATIONS = (
"CREATE TABLE IF NOT EXISTS lists (id TEXT PRIMARY KEY, board_id TEXT NOT NULL, "
"name TEXT NOT NULL, position DOUBLE PRECISION NOT NULL DEFAULT 0)",
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS list_id TEXT NOT NULL DEFAULT ''",
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS description TEXT NOT NULL DEFAULT ''",
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS labels TEXT NOT NULL DEFAULT ''",
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS position DOUBLE PRECISION NOT NULL DEFAULT 0",
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS blocked_by TEXT NOT NULL DEFAULT ''",
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS list_name TEXT",
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS embedding TEXT NOT NULL DEFAULT ''",
"CREATE TABLE IF NOT EXISTS comments (id TEXT PRIMARY KEY, card_id TEXT NOT NULL, "
"author TEXT NOT NULL, body TEXT NOT NULL, event_at TIMESTAMP)",
)
_schema_ready = False
def ensure_schema() -> None:
"""Apply _MIGRATIONS once per process, one statement per connection, never raising."""
global _schema_ready
if _schema_ready:
return
_schema_ready = True
for stmt in _MIGRATIONS:
try:
with connect() as c:
c.execute(stmt)
except Exception as exc: # already applied, or not applicable to this bucket's shape
print(f"schema step skipped ({stmt.split('(')[0].strip()}): {exc}",
file=sys.stderr)
def _card_from(row) -> Card:
return Card(id=row[0], board_id=row[1], list_id=row[2], title=row[3],
description=row[4] or "", assignee=row[5], due_date=row[6],
labels=tuple(x for x in (row[7] or "").split(",") if x),
position=float(row[8] or 0),
blocked_by=tuple(x for x in (row[9] or "").split(",") if x))
class SqlBoards:
def __init__(self) -> None:
ensure_schema()
def get(self, board_id: str) -> Optional[Board]:
with connect() 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, board_id, name, position 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(id=l[0], board_id=l[1], name=l[2], position=float(l[3]))
for l in lists])
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.get(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))
keep = {f"{board.id}:{e}" for e in board.members}
rows = c.execute("SELECT id FROM memberships WHERE board_id = %s",
(board.id,)).fetchall()
for (mid,) in rows:
if mid not in keep:
c.execute("DELETE FROM memberships WHERE id = %s", (mid,))
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))
def delete_board(self, board_id: str) -> None:
"""Remove the board and everything that hangs off it, in dependency
order: comments (through their cards), cards, lists, memberships,
activity, then the board row itself."""
with connect() as c:
c.execute("DELETE FROM comments WHERE card_id IN "
"(SELECT id FROM cards WHERE board_id = %s)", (board_id,))
c.execute("DELETE FROM cards WHERE board_id = %s", (board_id,))
c.execute("DELETE FROM lists WHERE board_id = %s", (board_id,))
c.execute("DELETE FROM memberships WHERE board_id = %s", (board_id,))
c.execute("DELETE FROM activity WHERE board_id = %s", (board_id,))
c.execute("DELETE FROM boards WHERE id = %s", (board_id,))
def save_lists(self, board: Board) -> None:
with connect() as c:
keep = {l.id for l in board.lists}
rows = c.execute("SELECT id FROM lists WHERE board_id = %s",
(board.id,)).fetchall()
for (lid,) in rows:
if lid not in keep:
c.execute("DELETE FROM lists WHERE id = %s", (lid,))
for lst in board.lists:
c.execute("INSERT INTO lists (id, board_id, name, position) "
"VALUES (%s, %s, %s, %s) ON CONFLICT (id) DO UPDATE SET "
"name = EXCLUDED.name, position = EXCLUDED.position",
(lst.id, lst.board_id, lst.name, lst.position))
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_from(r) if r else None
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 "
"ORDER BY position, updated_at", (board_id,)).fetchall()
return [_card_from(r) for r in rows]
def save_card(self, card: Card) -> None:
with connect() as c:
# Older buckets (remixed from the starter) carry a NOT NULL list_name column the
# current schema replaced with list_id; DataK3 cannot drop that constraint
# (ALTER COLUMN DROP NOT NULL is unsupported), so the name is written too,
# denormalized from the lists table. Fresh buckets have the column as well
# (schema.sql), so one code path serves both shapes.
row = c.execute("SELECT name FROM lists WHERE id = %s",
(card.list_id,)).fetchone()
list_name = row[0] if row else ""
c.execute("INSERT INTO cards (id, board_id, list_id, list_name, title, "
"description, assignee, due_date, labels, position, blocked_by, "
"updated_at) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s::date, %s, %s, %s, now()) "
"ON CONFLICT (id) DO UPDATE SET list_id = EXCLUDED.list_id, "
"list_name = EXCLUDED.list_name, "
"title = EXCLUDED.title, description = EXCLUDED.description, "
"assignee = EXCLUDED.assignee, due_date = EXCLUDED.due_date, "
"labels = EXCLUDED.labels, position = EXCLUDED.position, "
"blocked_by = EXCLUDED.blocked_by, updated_at = EXCLUDED.updated_at",
(card.id, card.board_id, card.list_id, list_name, card.title,
card.description, card.assignee, card.due_date,
",".join(card.labels), card.position, ",".join(card.blocked_by)))
def delete_card(self, card_id: str) -> None:
with connect() as c:
c.execute("DELETE FROM comments WHERE card_id = %s", (card_id,))
c.execute("DELETE FROM cards WHERE id = %s", (card_id,))
def save_comment(self, comment: Comment) -> None:
with connect() as c:
c.execute("INSERT INTO comments (id, card_id, author, body, event_at) "
"VALUES (%s, %s, %s, %s, now())",
(comment.id, comment.card_id, comment.author, comment.body))
def comment_counts(self, board_id: str) -> dict:
with connect() as c:
rows = c.execute("SELECT card_id, COUNT(*) FROM comments WHERE card_id IN "
"(SELECT id FROM cards WHERE board_id = %s) "
"GROUP BY card_id", (board_id,)).fetchall()
return {cid: int(n) for cid, n in rows}
def comments_on(self, card_id: str) -> list:
with connect() as c:
rows = c.execute("SELECT id, author, body, event_at FROM comments "
"WHERE card_id = %s ORDER BY event_at, id",
(card_id,)).fetchall()
return [Comment(id=r[0], card_id=card_id, author=r[1], body=r[2],
at=r[3].isoformat() if r[3] else None) for r in rows]
def get_embeddings(self, board_id: str) -> dict:
"""{card_id: (text_hash, vector)} for cards that carry a stored embedding.
The stored form is JSON {"h": hash of the card text it was made from,
"v": [floats]}; the hash lets the use case ignore vectors whose card text
has changed since they were computed."""
with connect() as c:
rows = c.execute("SELECT id, embedding FROM cards WHERE board_id = %s "
"AND embedding <> ''", (board_id,)).fetchall()
out = {}
for cid, raw in rows:
try:
d = json.loads(raw)
out[cid] = (d["h"], tuple(float(x) for x in d["v"]))
except Exception:
continue # a damaged value is treated as missing and recomputed
return out
def save_embedding(self, card_id: str, text_hash: str, vector) -> None:
raw = json.dumps({"h": text_hash, "v": [round(float(x), 5) for x in vector]})
with connect() as c:
c.execute("UPDATE cards SET embedding = %s WHERE id = %s", (raw, card_id))
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))
def activity(self, board_id: str, limit: int = 50) -> list:
with connect() as c:
rows = c.execute("SELECT actor, action, card_id, event_at FROM activity "
"WHERE board_id = %s ORDER BY event_at DESC LIMIT %s",
(board_id, limit)).fetchall()
return [{"actor": a, "action": act, "card_id": cid,
"at": ts.isoformat() if ts else None} for a, act, cid, ts in rows]
def counts(self) -> dict:
with connect() as c:
out = {}
for table in ("boards", "lists", "cards", "memberships", "activity", "comments"):
out[table] = c.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
return out