dachdigitalwerksorg/kanvaflow
KanvaFlow — Remixable team task board with RBAC, drag-and-drop Kanban, dependencies, Dodil Models AI, and MCP tools.
A remix template · 0 remixes · Remix dachdigitalwerksorg/kanvaflow
app/boards/infrastructure/repository.py
"""DataK3 SQL Repository (`schema.sql`).
Uses natural primary keys and idempotent `INSERT ... ON CONFLICT (id) DO UPDATE` upserts.
Also runs a lightweight additive schema check once per process so existing buckets upgraded
from the starter schema gain the extended KanvaFlow columns and tables automatically.
"""
import json
import re
import threading
import uuid
from typing import Optional
from boards.domain.model import Board, BoardList, Card, Role, default_board_lists
from boards.infrastructure.dodil import connect
def _slug(name: str) -> str:
s = re.sub(r"[^a-z0-9]+", "-", name.strip().lower()).strip("-")
return s or uuid.uuid4().hex[:8]
def _loads_list(raw: Optional[str]) -> list:
if not raw:
return []
try:
v = json.loads(raw)
return v if isinstance(v, list) else []
except Exception:
return []
class SqlBoards:
def __init__(self) -> None:
self._migrated = False
self._mig_lock = threading.Lock()
def _conn(self):
c = connect()
if not self._migrated:
with self._mig_lock:
if not self._migrated:
self._ensure_schema(c)
self._migrated = True
return c
@staticmethod
def _ensure_schema(c) -> None:
ddl = [
"CREATE TABLE IF NOT EXISTS board_lists (id TEXT PRIMARY KEY, board_id TEXT NOT NULL, name TEXT NOT NULL, position INTEGER NOT NULL, is_done BOOLEAN NOT NULL)",
"CREATE TABLE IF NOT EXISTS card_dependencies (id TEXT PRIMARY KEY, board_id TEXT NOT NULL, card_id TEXT NOT NULL, blocked_by_id TEXT NOT NULL)",
"ALTER TABLE boards ADD COLUMN IF NOT EXISTS description TEXT",
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS description TEXT",
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS labels_json TEXT",
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS checklist_json TEXT",
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS embedding_json TEXT",
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS position INTEGER",
]
for stmt in ddl:
try:
c.execute(stmt)
except Exception:
pass
def get(self, board_id: str) -> Optional[Board]:
with self._conn() as c:
row = c.execute(
"SELECT id, title, owner_email, COALESCE(description, '') 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()
list_rows = c.execute(
"SELECT name, position, is_done FROM board_lists WHERE board_id = %s ORDER BY position",
(board_id,),
).fetchall()
lists = (
[BoardList(name=r[0], position=int(r[1]), is_done=bool(r[2])) for r in list_rows]
if list_rows
else default_board_lists()
)
return Board(
id=row[0],
title=row[1],
owner_email=row[2],
description=row[3] or "",
members={e: Role(r) for e, r in members if e != row[2]},
lists=lists,
)
def boards_for(self, email: str) -> list[Board]:
clean = email.strip().lower()
with self._conn() 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",
(clean, clean),
).fetchall()
]
return [b for b in (self.get(i) for i in ids) if b is not None]
def save_board(self, board: Board) -> None:
with self._conn() as c:
c.execute(
"INSERT INTO boards (id, title, description, owner_email, created_at) "
"VALUES (%s, %s, %s, %s, now()) ON CONFLICT (id) DO UPDATE SET "
"title = EXCLUDED.title, description = EXCLUDED.description, owner_email = EXCLUDED.owner_email",
(board.id, board.title, board.description, 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),
)
# Sync board_lists cleanly
keep_ids = []
for lst in board.lists:
lid = f"{board.id}:{_slug(lst.name)}"
keep_ids.append(lid)
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",
(lid, board.id, lst.name, lst.position, lst.is_done),
)
existing_lids = [
r[0]
for r in c.execute("SELECT id FROM board_lists WHERE board_id = %s", (board.id,)).fetchall()
]
for old_lid in existing_lids:
if old_lid not in keep_ids:
c.execute("DELETE FROM board_lists WHERE id = %s", (old_lid,))
def remove_member_row(self, board_id: str, email: str) -> None:
with self._conn() as c:
c.execute("DELETE FROM memberships WHERE id = %s", (f"{board_id}:{email}",))
@staticmethod
def _row_to_card(r) -> Card:
emb_raw = _loads_list(r[8])
emb = [float(x) for x in emb_raw] if emb_raw else None
return Card(
id=r[0],
board_id=r[1],
title=r[2],
list_name=r[3],
description=r[4] or "",
assignee=r[5],
due_date=r[6],
labels=[str(x) for x in _loads_list(r[7])],
checklist=_loads_list(r[9]),
position=int(r[10]) if r[10] is not None else 0,
embedding=emb,
)
def get_card(self, card_id: str) -> Optional[Card]:
with self._conn() as c:
r = c.execute(
"SELECT id, board_id, title, list_name, COALESCE(description, ''), assignee, due_date, "
"labels_json, embedding_json, checklist_json, COALESCE(position, 0) "
"FROM cards WHERE id = %s",
(card_id,),
).fetchone()
return self._row_to_card(r) if r else None
def cards_on(self, board_id: str) -> list[Card]:
with self._conn() as c:
rows = c.execute(
"SELECT id, board_id, title, list_name, COALESCE(description, ''), assignee, due_date, "
"labels_json, embedding_json, checklist_json, COALESCE(position, 0) "
"FROM cards WHERE board_id = %s ORDER BY COALESCE(position, 0), updated_at",
(board_id,),
).fetchall()
return [self._row_to_card(r) for r in rows]
def save_card(self, card: Card) -> None:
labels_s = json.dumps(card.labels)
checklist_s = json.dumps(card.checklist)
emb_s = json.dumps(card.embedding) if card.embedding else None
with self._conn() as c:
c.execute(
"INSERT INTO cards (id, board_id, list_name, title, description, assignee, due_date, "
"labels_json, checklist_json, embedding_json, position, updated_at) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, now()) "
"ON CONFLICT (id) DO UPDATE SET list_name = EXCLUDED.list_name, "
"title = EXCLUDED.title, description = EXCLUDED.description, "
"assignee = EXCLUDED.assignee, due_date = EXCLUDED.due_date, "
"labels_json = EXCLUDED.labels_json, checklist_json = EXCLUDED.checklist_json, "
"embedding_json = COALESCE(EXCLUDED.embedding_json, cards.embedding_json), "
"position = EXCLUDED.position, updated_at = EXCLUDED.updated_at",
(
card.id,
card.board_id,
card.list_name,
card.title,
card.description,
card.assignee,
card.due_date,
labels_s,
checklist_s,
emb_s,
card.position,
),
)
def delete_card(self, card_id: str) -> None:
with self._conn() as c:
c.execute("DELETE FROM card_dependencies WHERE card_id = %s OR blocked_by_id = %s", (card_id, card_id))
c.execute("DELETE FROM cards WHERE id = %s", (card_id,))
def dependencies_on(self, board_id: str) -> dict[str, list[str]]:
with self._conn() as c:
rows = c.execute(
"SELECT card_id, blocked_by_id FROM card_dependencies WHERE board_id = %s",
(board_id,),
).fetchall()
out: dict[str, list[str]] = {}
for cid, bid in rows:
out.setdefault(cid, []).append(bid)
return out
def add_dependency(self, board_id: str, card_id: str, blocked_by_id: str) -> None:
with self._conn() as c:
c.execute(
"INSERT INTO card_dependencies (id, board_id, card_id, blocked_by_id) "
"VALUES (%s, %s, %s, %s) ON CONFLICT (id) DO NOTHING",
(f"{card_id}:{blocked_by_id}", board_id, card_id, blocked_by_id),
)
def remove_dependency(self, card_id: str, blocked_by_id: str) -> None:
with self._conn() as c:
c.execute("DELETE FROM card_dependencies WHERE id = %s", (f"{card_id}:{blocked_by_id}",))
def log(self, board_id: str, card_id: Optional[str], actor: str, action: str) -> None:
with self._conn() 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.strip().lower(), action),
)
def activity_for_board(self, board_id: str, limit: int = 40) -> list[dict]:
with self._conn() as c:
rows = c.execute(
"SELECT id, card_id, actor, action, event_at FROM activity "
"WHERE board_id = %s ORDER BY event_at DESC LIMIT %s",
(board_id, limit),
).fetchall()
return [
{
"id": r[0],
"card_id": r[1],
"actor": r[2],
"action": r[3],
"event_at": r[4].isoformat() if hasattr(r[4], "isoformat") else str(r[4] or ""),
}
for r in rows
]
def activity_for_card(self, card_id: str, limit: int = 40) -> list[dict]:
with self._conn() as c:
rows = c.execute(
"SELECT id, card_id, actor, action, event_at FROM activity "
"WHERE card_id = %s ORDER BY event_at DESC LIMIT %s",
(card_id, limit),
).fetchall()
return [
{
"id": r[0],
"card_id": r[1],
"actor": r[2],
"action": r[3],
"event_at": r[4].isoformat() if hasattr(r[4], "isoformat") else str(r[4] or ""),
}
for r in rows
]
def health_counts(self) -> dict[str, int]:
counts = {}
with self._conn() as c:
for tbl in ("boards", "memberships", "board_lists", "cards", "card_dependencies", "activity"):
row = c.execute(f"SELECT count(*) FROM {tbl}").fetchone()
counts[tbl] = int(row[0]) if row else 0
return counts