ghpsolutions/quadro
Quadro: team boards with roles, history and an MCP endpoint for Claude and ChatGPT
A remix template · 0 remixes · Remix ghpsolutions/quadro
app/boards/infrastructure/repository.py
"""The DataK3 repository: boards, cards and history in the app's bucket, over the Postgres wire.
DataK3 notes that shaped this file:
- no sequences, so card ids are "<board>-<number>" claimed with a plain INSERT (23505 = taken);
- upserts rewrite every column they are given, so they always carry the whole row;
- a few SQLSTATEs are transient on the platform and are retried;
- edits use compare-and-set on row_version, so a stale edit fails instead of overwriting.
"""
from __future__ import annotations
import json
import os
import queue
import threading
import time
from contextlib import contextmanager
from pathlib import Path
from typing import Optional, Sequence
import psycopg
from psycopg import errors
from boards.application.ports import NumberTaken
from boards.domain.model import Board, BoardList, Card, Role
from boards.infrastructure.dodil import connect
RETRY_STATES = {"40001", "08006", "58030", "XX000"}
SCHEMA = Path(__file__).with_name("schema.sql")
MAX_AGE = 240.0 # seconds; the connection password is a short-lived token
CARD_COLUMNS = ("id, board_id, card_number, list_id, title, description, assignee, due_date, "
"labels, blocked_by, sort_order, row_version, archived, created_by, created_at, "
"updated_at")
class _Pool:
"""A few reused connections. Opening a TLS connection per query is the slow part."""
def __init__(self, size: int):
self.idle, self.gate = queue.LifoQueue(), threading.BoundedSemaphore(size)
@contextmanager
def conn(self):
with self.gate:
item = None
while item is None:
try:
c, born = self.idle.get_nowait()
except queue.Empty:
c, born = connect(), time.time()
if c.closed or time.time() - born > MAX_AGE:
c.close()
continue
item = (c, born)
ok = False
try:
yield item[0]
ok = True
finally:
if ok and not item[0].closed:
self.idle.put(item)
else:
item[0].close()
class SqlBoards:
def __init__(self, pool_size: Optional[int] = None):
self.pool = _Pool(pool_size or int(os.environ.get("DB_POOL", "8")))
# -------------------------------------------------------------- plumbing
def _run(self, sql: str, args: tuple = (), fetch: bool = False):
for attempt in range(4):
try:
with self.pool.conn() as c, c.cursor() as cur:
cur.execute(sql, args)
if fetch:
return cur.fetchall()
return cur.rowcount
except errors.UniqueViolation:
raise
except psycopg.Error as e:
transient = isinstance(e, psycopg.OperationalError) or e.sqlstate in RETRY_STATES
if not transient or attempt == 3:
raise
time.sleep(0.2 * (2 ** attempt))
def ensure_schema(self) -> None:
for statement in SCHEMA.read_text().split(";"):
body = "\n".join(l for l in statement.splitlines() if not l.strip().startswith("--"))
if body.strip():
self._run(body)
# -------------------------------------------------------------- boards and people
def save_board(self, board: Board) -> None:
self._run("INSERT INTO boards (id, board_key, title, owner_email, created_at) "
"VALUES (%s, %s, %s, %s, %s) ON CONFLICT (id) DO UPDATE SET "
"board_key = EXCLUDED.board_key, title = EXCLUDED.title, "
"owner_email = EXCLUDED.owner_email, created_at = EXCLUDED.created_at",
(board.id, board.key, board.title, board.owner, board.created_at))
def _boards(self, ids: Sequence[str]) -> list:
if not ids:
return []
rows = self._run("SELECT id, board_key, title, owner_email, created_at FROM boards "
"WHERE id = ANY(%s)", (list(ids),), fetch=True)
members = self._run("SELECT board_id, email, role FROM memberships WHERE board_id = ANY(%s)",
(list(ids),), fetch=True)
boards = {r[0]: Board(id=r[0], key=r[1], title=r[2], owner=r[3], created_at=r[4])
for r in rows}
for bid, email, role in members:
if bid in boards and role != Role.OWNER.value:
boards[bid].members[email] = Role(role)
return list(boards.values())
def get_board(self, board_id: str) -> Optional[Board]:
found = self._boards([board_id])
return found[0] if found else None
def boards_for(self, email: str) -> list:
rows = self._run("SELECT board_id FROM memberships WHERE email = %s", (email,), fetch=True)
return self._boards([r[0] for r in rows])
def save_member(self, board_id: str, email: str, role: Role) -> None:
self._run("INSERT INTO memberships (id, board_id, email, role) VALUES (%s, %s, %s, %s) "
"ON CONFLICT (id) DO UPDATE SET board_id = EXCLUDED.board_id, "
"email = EXCLUDED.email, role = EXCLUDED.role",
(f"{board_id}:{email}", board_id, email, role.value))
def remove_member(self, board_id: str, email: str) -> None:
self._run("DELETE FROM memberships WHERE id = %s", (f"{board_id}:{email}",))
# -------------------------------------------------------------- lists
def save_list(self, lst: BoardList) -> None:
self._run("INSERT INTO board_lists (id, board_id, name, sort_order, is_done) "
"VALUES (%s, %s, %s, %s, %s) ON CONFLICT (id) DO UPDATE SET "
"board_id = EXCLUDED.board_id, name = EXCLUDED.name, "
"sort_order = EXCLUDED.sort_order, is_done = EXCLUDED.is_done",
(lst.id, lst.board_id, lst.name, lst.position, lst.done))
def lists_on(self, board_id: str) -> list:
rows = self._run("SELECT id, board_id, name, sort_order, is_done FROM board_lists "
"WHERE board_id = %s", (board_id,), fetch=True)
return [BoardList(r[0], r[1], r[2], float(r[3]), bool(r[4])) for r in rows]
def delete_list(self, list_id: str) -> None:
self._run("DELETE FROM board_lists WHERE id = %s", (list_id,))
# -------------------------------------------------------------- cards
@staticmethod
def _card(r) -> Card:
return Card(id=r[0], board_id=r[1], number=int(r[2]), list_id=r[3], title=r[4],
description=r[5] or "", assignee=r[6], due_date=r[7],
labels=tuple(json.loads(r[8] or "[]")),
blocked_by=tuple(json.loads(r[9] or "[]")), position=float(r[10]),
version=int(r[11]), created_by=r[13], created_at=r[14], updated_at=r[15])
@staticmethod
def _values(card: Card, archived: bool = False) -> tuple:
return (card.id, card.board_id, card.number, card.list_id, card.title, card.description,
card.assignee, card.due_date, json.dumps(list(card.labels)),
json.dumps(list(card.blocked_by)), card.position, card.version, archived,
card.created_by, card.created_at, card.updated_at)
def next_number(self, board_id: str) -> int:
rows = self._run("SELECT MAX(card_number) FROM cards WHERE board_id = %s", (board_id,),
fetch=True)
return int(rows[0][0] or 0) + 1 if rows else 1
def insert_card(self, card: Card) -> None:
try:
self._run(f"INSERT INTO cards ({CARD_COLUMNS}) VALUES ({', '.join(['%s'] * 16)})",
self._values(card))
except errors.UniqueViolation:
raise NumberTaken(card.id)
def update_card(self, card: Card, expected_version: int) -> bool:
count = self._run(
"UPDATE cards SET list_id = %s, title = %s, description = %s, assignee = %s, "
"due_date = %s, labels = %s, blocked_by = %s, sort_order = %s, row_version = %s, "
"updated_at = %s WHERE id = %s AND row_version = %s AND archived = FALSE",
(card.list_id, card.title, card.description, card.assignee, card.due_date,
json.dumps(list(card.labels)), json.dumps(list(card.blocked_by)), card.position,
card.version, card.updated_at, card.id, expected_version))
if count is not None and count >= 0:
return count > 0
stored = self.get_card(card.id) # driver could not report a row count
return bool(stored and stored.version == card.version and stored.updated_at == card.updated_at)
def get_card(self, card_id: str) -> Optional[Card]:
rows = self._run(f"SELECT {CARD_COLUMNS} FROM cards WHERE id = %s AND archived = FALSE",
(card_id,), fetch=True)
return self._card(rows[0]) if rows else None
def cards_on(self, board_ids: Sequence[str]) -> list:
if not board_ids:
return []
rows = self._run(f"SELECT {CARD_COLUMNS} FROM cards WHERE board_id = ANY(%s) "
"AND archived = FALSE", (list(board_ids),), fetch=True)
return [self._card(r) for r in rows]
def delete_card(self, card_id: str) -> None:
# Archived, not erased: the number stays taken and the history keeps its subject.
self._run("UPDATE cards SET archived = TRUE WHERE id = %s", (card_id,))
# -------------------------------------------------------------- comments and history
def add_comment(self, c: dict) -> None:
self._run("INSERT INTO comments (id, card_id, board_id, author, body, created_at) "
"VALUES (%s, %s, %s, %s, %s, %s)",
(c["id"], c["card_id"], c["board_id"], c["author"], c["body"], c["created_at"]))
def comments_for(self, card_id: str) -> list:
rows = self._run("SELECT id, card_id, board_id, author, body, created_at FROM comments "
"WHERE card_id = %s ORDER BY created_at", (card_id,), fetch=True)
keys = ("id", "card_id", "board_id", "author", "body", "created_at")
return [dict(zip(keys, r)) for r in rows]
def comment_counts(self, board_id: str) -> dict:
rows = self._run("SELECT card_id, COUNT(*) AS n FROM comments WHERE board_id = %s "
"GROUP BY card_id", (board_id,), fetch=True)
return {r[0]: int(r[1]) for r in rows}
def record(self, e: dict) -> None:
self._run("INSERT INTO activity (id, board_id, card_id, card_ref, actor, action, changes, "
"note, happened_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)",
(e["id"], e["board_id"], e["card_id"], e["card_ref"], e["actor"], e["action"],
json.dumps(e["changes"]), e["note"] or "", e["happened_at"]))
def activity(self, board_id: str, card_id: Optional[str] = None, limit: int = 50) -> list:
sql = ("SELECT id, board_id, card_id, card_ref, actor, action, changes, note, happened_at "
"FROM activity WHERE board_id = %s")
args = [board_id]
if card_id is not None:
sql += " AND card_id = %s"
args.append(card_id)
sql += " ORDER BY happened_at DESC, id DESC LIMIT %s"
args.append(int(limit))
keys = ("id", "board_id", "card_id", "card_ref", "actor", "action", "changes", "note",
"happened_at")
out = []
for r in self._run(sql, tuple(args), fetch=True):
row = dict(zip(keys, r))
row["changes"] = json.loads(row["changes"] or "[]")
out.append(row)
return out
# -------------------------------------------------------------- similarity index
def save_embedding(self, card_id: str, board_id: str, digest: str, vector: list) -> None:
self._run("INSERT INTO card_embeddings (card_id, board_id, digest, embedding) "
"VALUES (%s, %s, %s, %s) ON CONFLICT (card_id) DO UPDATE SET "
"board_id = EXCLUDED.board_id, digest = EXCLUDED.digest, "
"embedding = EXCLUDED.embedding",
(card_id, board_id, digest, json.dumps([round(float(x), 6) for x in vector])))
def embeddings_on(self, board_id: str) -> dict:
rows = self._run("SELECT card_id, digest, embedding FROM card_embeddings "
"WHERE board_id = %s", (board_id,), fetch=True)
return {r[0]: (r[1], json.loads(r[2])) for r in rows}