flc/crewboard
Crewboard - team task board (boards, lists, cards, roles, history, drag and drop, MCP, AI extras). FastAPI + DataK3 + appid + Models.
A remix template · 1 remixes · Remix flc/crewboard
app/boards/infrastructure/repository.py
"""Boards and cards stored in DataK3 tables (schema.sql).
- New keys (board keys, card numbers, activity ids) are claimed by a PLAIN INSERT: under a race one
writer wins and the rest get 23505, which becomes Taken and the use case tries the next key.
- A card update is `UPDATE ... WHERE id AND version`: a stale write changes no row (no lost
updates), and only the columns named in SET are touched.
- Lists, labels and dependencies are small JSON arrays in TEXT columns.
"""
import json
import uuid
from typing import Optional
import psycopg
from boards.application.ports import Taken
from boards.domain.model import Board, Card, Role
from boards.infrastructure.dodil import connect
CARD_COLS = ("id, board_id, number, title, list_name, position, description, assignee, "
"due_date, labels, blocked_by, archived, version")
def _date(v):
return v.date() if hasattr(v, "date") and callable(v.date) else v
def _card(r) -> Card:
return Card(id=r[0], board_id=r[1], number=int(r[2] or 0), title=r[3], list_name=r[4],
position=float(r[5] or 0), description=r[6] or "", assignee=r[7],
due_date=_date(r[8]), labels=json.loads(r[9] or "[]"),
blocked_by=json.loads(r[10] or "[]"), archived=bool(r[11]),
version=int(r[12] or 1))
class SqlBoards:
def get(self, board_id: str) -> Optional[Board]:
with connect() as c:
row = c.execute("SELECT id, title, owner_email, lists 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()
board = 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})
if row[3]:
board.lists = json.loads(row[3])
return board
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 sorted((b for b in (self.get(i) for i in ids) if b), key=lambda b: b.title.lower())
def insert_board(self, board: Board) -> None:
try:
with connect() as c:
c.execute("INSERT INTO boards (id, title, owner_email, lists, created_at) "
"VALUES (%s, %s, %s, %s, now())",
(board.id, board.title, board.owner_email, json.dumps(board.lists)))
except psycopg.errors.UniqueViolation:
raise Taken(board.id)
def save_board(self, board: Board) -> None:
with connect() as c:
c.execute("UPDATE boards SET title = %s, lists = %s WHERE id = %s",
(board.title, json.dumps(board.lists), board.id))
def save_members(self, board: Board, removed: tuple = ()) -> None:
with connect() as c:
for email in removed:
c.execute("DELETE FROM memberships WHERE id = %s", (f"{board.id}:{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))
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 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", (board_id,)).fetchall()
return [_card(r) for r in rows]
def cards_on_boards(self, board_ids: list) -> list:
return [card for b in board_ids for card in self.cards_on(b)]
def max_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) if r else 0
def insert_card(self, card: Card) -> None:
try:
with connect() as c:
c.execute(f"INSERT INTO cards ({CARD_COLS}, created_at, updated_at) VALUES "
"(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, now(), now())",
self._values(card))
except psycopg.errors.UniqueViolation:
raise Taken(card.id)
def update_card(self, card: Card, expected_version: int) -> bool:
with connect() as c:
cur = c.execute(
"UPDATE cards SET title = %s, list_name = %s, position = %s, description = %s, "
"assignee = %s, due_date = %s, labels = %s, blocked_by = %s, archived = %s, "
"version = %s, updated_at = now() WHERE id = %s AND version = %s",
(card.title, card.list_name, card.position, card.description, card.assignee,
card.due_date, json.dumps(card.labels), json.dumps(card.blocked_by),
card.archived, card.version, card.id, expected_version))
return cur.rowcount == 1
@staticmethod
def _values(card: Card) -> tuple:
return (card.id, card.board_id, card.number, card.title, card.list_name, card.position,
card.description, card.assignee, card.due_date, json.dumps(card.labels),
json.dumps(card.blocked_by), card.archived, card.version)
def log(self, board_id: str, card_id: Optional[str], actor: str, action: str,
changes: Optional[dict] = None) -> None:
with connect() as c:
c.execute("INSERT INTO activity (id, board_id, card_id, actor, action, changes, "
"event_at) VALUES (%s, %s, %s, %s, %s, %s, now())",
(uuid.uuid4().hex, board_id, card_id, actor, action,
json.dumps(changes, default=str) if changes else None))
def activity(self, board_id: str, card_id: Optional[str] = None, limit: int = 50) -> list:
sql = ("SELECT card_id, actor, action, changes, event_at FROM activity "
"WHERE board_id = %s")
args = [board_id]
if card_id:
sql += " AND card_id = %s"
args.append(card_id)
sql += " ORDER BY event_at DESC LIMIT %s"
args.append(int(limit))
with connect() as c:
rows = c.execute(sql, args).fetchall()
return [{"card_id": r[0], "actor": r[1], "action": r[2],
"changes": json.loads(r[3]) if r[3] else None,
"at": r[4].isoformat() if hasattr(r[4], "isoformat") else r[4]} for r in rows]
def counts(self) -> dict:
"""Row count per table, read with the app's own credential (the POST / self-check)."""
with connect() as c:
return {t: c.execute(f"SELECT count(*) FROM {t}").fetchone()[0]
for t in ("boards", "memberships", "cards", "activity")}