rahulpawar/taskdeck
Lanebook, a team task board (boards, lists, cards, roles, history) you run from the web or from Claude/ChatGPT over MCP. FastAPI + DataK3 + appid + Models.
A remix template · 0 remixes · Remix rahulpawar/taskdeck
app/boards/infrastructure/repository.py
"""Boards, lists, cards, people and history stored in DataK3 tables (schema.sql).
Writes are keyed upserts (INSERT ... ON CONFLICT (id) DO UPDATE), so a retry is safe. DataK3 has
no sequences, so every key is chosen here: uuids, board_id:email for memberships, and a
per-board card number computed as MAX + 1.
"""
import uuid
from typing import Optional
from boards.domain.model import Board, BoardList, Card, Role
from boards.infrastructure.dodil import connect
CARD_COLS = ("id, board_id, number, list_id, title, description, assignee, due_date, labels, "
"position, blocked_by")
def _card(r) -> Card:
return Card(id=r[0], board_id=r[1], number=int(r[2] or 0), list_id=r[3], title=r[4],
description=r[5] or "", assignee=r[6], due_date=r[7],
labels=[x for x in (r[8] or "").split(",") if x], position=float(r[9] or 0),
blocked_by=[x for x in (r[10] or "").split(",") if x])
def _in(ids: list) -> tuple:
return ", ".join(["%s"] * len(ids)), tuple(ids)
class SqlBoards:
# ----- boards and people -----
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()
return Board(id=row[0], title=row[1], owner_email=row[2],
members={e: Role(r) for e, r in members if r != "owner"})
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))]
if not ids:
return []
ph, args = _in(ids)
rows = c.execute(f"SELECT id, title, owner_email FROM boards WHERE id IN ({ph})",
args).fetchall()
mem = c.execute(f"SELECT board_id, email, role FROM memberships "
f"WHERE board_id IN ({ph})", args).fetchall()
out = []
for bid, title, owner in rows:
members = {e: Role(r) for b, e, r in mem if b == bid and r != "owner"}
out.append(Board(id=bid, title=title, owner_email=owner, members=members))
return out
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))
def remove_membership(self, board_id: str, email: str) -> None:
with connect() as c:
c.execute("DELETE FROM memberships WHERE id = %s", (f"{board_id}:{email}",))
# ----- lists -----
def lists_on(self, board_id: str) -> list:
with connect() as c:
rows = c.execute("SELECT id, board_id, name, position FROM lists WHERE board_id = %s",
(board_id,)).fetchall()
return [BoardList(r[0], r[1], r[2], float(r[3] or 0)) for r in rows]
def save_list(self, lst: BoardList) -> None:
with connect() as c:
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 delete_list(self, list_id: str) -> None:
with connect() as c:
c.execute("DELETE FROM lists WHERE id = %s", (list_id,))
# ----- cards -----
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_ids: list) -> list:
if not board_ids:
return []
ph, args = _in(list(board_ids))
with connect() as c:
rows = c.execute(f"SELECT {CARD_COLS} FROM cards WHERE board_id IN ({ph})",
args).fetchall()
return [_card(r) for r in rows]
def save_card(self, card: Card) -> None:
with connect() 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 (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, "
"position = EXCLUDED.position, blocked_by = EXCLUDED.blocked_by, "
"updated_at = EXCLUDED.updated_at",
(card.id, card.board_id, card.number, card.list_id, 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 cards WHERE id = %s", (card_id,))
def next_card_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] if r else None) or 0) + 1
# ----- history -----
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 history(self, board_id: str, card_id: Optional[str] = None, limit: int = 50) -> list:
sql = "SELECT actor, action, event_at, card_id FROM activity WHERE board_id = %s"
args = [board_id]
if card_id:
sql += " AND card_id = %s"
args.append(card_id)
sql += f" ORDER BY event_at DESC LIMIT {int(limit)}"
with connect() as c:
rows = c.execute(sql, tuple(args)).fetchall()
return [{"actor": a, "action": t, "at": e.isoformat() if hasattr(e, "isoformat") else str(e),
"card_id": cid} for a, t, e, cid in rows]