stubborn-otter/otter-boards
Otter Boards, a team task board you can run from the web or from Claude and ChatGPT. FastAPI + DataK3 + appid + Models + MCP.
A remix template · 1 remixes · Remix stubborn-otter/otter-boards
app/boards/infrastructure/repository.py
"""Boards and cards stored in DataK3 tables (schema.sql).
DataK3 speaks the Postgres wire protocol but is not full Postgres, so the SQL stays deliberately
small: SELECT with WHERE, ORDER BY and LIMIT, and INSERT. Writes are keyed upserts, so a retry is
safe: INSERT ... ON CONFLICT (id) DO UPDATE SET col = EXCLUDED.col. There is no UPDATE or DELETE;
what leaves the board is marked archived or removed. The one plain INSERT that may fail on purpose
is insert_card, which is how card numbers stay unique without a sequence.
"""
import json
import logging
import threading
import time
import uuid
from datetime import datetime, timezone
from typing import Callable, Optional
import psycopg
from boards.domain.model import Board, BoardList, Card, ChecklistItem, Event, Role
from boards.infrastructure.dodil import connect
REMOVED = "removed" # a memberships.role value: the row stays, the person doesn't
MAX_CONNECTION_AGE = 240 # seconds. The service-account token that opened it is short-lived.
RETRY_PAUSES = (0.3, 1, 2, 4, 8, 16) # seconds between tries: about 30 in all, to outlast DataK3 waking up
log = logging.getLogger("otter-boards.repository")
_CARD = ("id, board_id, card_number, title, list_id, sort_order, description, assignee, due_date, "
"labels, checklist, blocked_by, archived")
_LIST = "id, board_id, name, sort_order, is_done, archived"
_EVENT = "id, board_id, card_id, actor, action, event_at, field_name, old_value, new_value"
def _card(r) -> Card:
return Card(id=r[0], board_id=r[1], number=r[2], title=r[3], list_id=r[4], position=r[5],
description=r[6], assignee=r[7], due_date=r[8], labels=json.loads(r[9]),
checklist=[ChecklistItem(i["text"], i["done"]) for i in json.loads(r[10])],
blocked_by=json.loads(r[11]), archived=r[12])
def _card_row(c: Card) -> tuple:
return (c.id, c.board_id, c.number, c.title, c.list_id, c.position, c.description, c.assignee,
c.due_date, json.dumps(c.labels),
json.dumps([{"text": i.text, "done": i.done} for i in c.checklist]),
json.dumps(c.blocked_by), c.archived)
def _utc(at: datetime) -> datetime:
"""Times are UTC without a zone in the app. DataK3 hands TIMESTAMP back with one."""
return at.astimezone(timezone.utc).replace(tzinfo=None) if at.tzinfo else at
def _event(r) -> Event:
return Event(id=r[0], board_id=r[1], card_id=r[2], actor=r[3], action=r[4], at=_utc(r[5]),
field=r[6], old=r[7], new=r[8])
class SqlBoards:
def __init__(self, open_connection: Callable[[], psycopg.Connection] = connect):
self._open, self._local = open_connection, threading.local()
def _connection(self) -> psycopg.Connection:
"""One connection per worker thread: opening one means TLS, a token and a trip to London,
and a single page load runs several queries."""
conn = getattr(self._local, "conn", None)
if conn is None or conn.closed or time.time() - self._local.opened > MAX_CONNECTION_AGE:
self._drop()
self._local.conn, self._local.opened = self._open(), time.time()
return self._local.conn
def _drop(self) -> None:
conn, self._local.conn = getattr(self._local, "conn", None), None
if conn is not None:
try:
conn.close()
except psycopg.Error:
pass
def _sql(self, statement: str, params: tuple = ()) -> list:
"""Run one statement. DataK3 is slow to answer after a quiet spell and refuses meanwhile
("tcp connect error"), so a failed connection is replaced and tried again after a pause."""
for pause in (*RETRY_PAUSES, None):
try:
cursor = self._connection().execute(statement, params)
return cursor.fetchall() if cursor.description else []
except psycopg.OperationalError:
self._drop()
if pause is None:
raise
time.sleep(pause)
return []
def get(self, board_id: str) -> Optional[Board]:
rows = self._sql("SELECT id, title, owner_email FROM boards WHERE id = %s", (board_id,))
if not rows:
return None
members = self._sql("SELECT email, role FROM memberships "
"WHERE board_id = %s AND role <> %s", (board_id, REMOVED))
row = rows[0]
return Board(id=row[0], title=row[1], owner_email=row[2],
members={e: Role(r) for e, r in members})
def boards_for(self, email: str) -> list:
owned = self._sql("SELECT id FROM boards WHERE owner_email = %s", (email,))
joined = self._sql("SELECT board_id FROM memberships WHERE email = %s AND role <> %s",
(email, REMOVED))
ids = dict.fromkeys(r[0] for r in owned + joined)
return [b for b in (self.get(i) for i in ids) if b]
def save_board(self, board: Board) -> None:
self._sql("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():
self._save_membership(board.id, email, role.value)
def _save_membership(self, board_id: str, email: str, role: str) -> None:
self._sql("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))
def remove_member(self, board_id: str, email: str) -> None:
self._save_membership(board_id, email, REMOVED)
def lists_on(self, board_id: str) -> list:
rows = self._sql(f"SELECT {_LIST} FROM lists WHERE board_id = %s AND archived = %s "
"ORDER BY sort_order", (board_id, False))
return [BoardList(*r) for r in rows]
def save_list(self, x: BoardList) -> None:
self._sql(f"INSERT INTO lists ({_LIST}) VALUES (%s, %s, %s, %s, %s, %s) "
"ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, "
"sort_order = EXCLUDED.sort_order, is_done = EXCLUDED.is_done, "
"archived = EXCLUDED.archived",
(x.id, x.board_id, x.name, x.position, x.is_done, x.archived))
def get_card(self, card_id: str) -> Optional[Card]:
rows = self._sql(f"SELECT {_CARD} FROM cards WHERE id = %s", (card_id,))
return _card(rows[0]) if rows else None
def cards_on(self, board_id: str) -> list:
rows = self._sql(f"SELECT {_CARD} FROM cards WHERE board_id = %s AND archived = %s "
"ORDER BY sort_order, card_number", (board_id, False))
return [_card(r) for r in rows]
def last_card_number(self, board_id: str) -> int:
rows = self._sql("SELECT card_number FROM cards WHERE board_id = %s "
"ORDER BY card_number DESC LIMIT 1", (board_id,))
return rows[0][0] if rows else 0
def insert_card(self, card: Card) -> bool:
try:
self._sql(f"INSERT INTO cards ({_CARD}, updated_at) VALUES "
"(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, now())",
_card_row(card))
return True
except psycopg.errors.UniqueViolation:
log.warning("card number already taken: %s", card.id)
return False
def save_card(self, card: Card) -> None:
self._sql(f"INSERT INTO cards ({_CARD}, updated_at) VALUES "
"(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, now()) "
"ON CONFLICT (id) DO UPDATE SET title = EXCLUDED.title, "
"list_id = EXCLUDED.list_id, sort_order = EXCLUDED.sort_order, "
"description = EXCLUDED.description, assignee = EXCLUDED.assignee, "
"due_date = EXCLUDED.due_date, labels = EXCLUDED.labels, "
"checklist = EXCLUDED.checklist, blocked_by = EXCLUDED.blocked_by, "
"archived = EXCLUDED.archived, updated_at = EXCLUDED.updated_at",
_card_row(card))
def log(self, e: Event) -> None:
self._sql(f"INSERT INTO activity ({_EVENT}) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)",
(e.id or uuid.uuid4().hex, e.board_id, e.card_id, e.actor, e.action, e.at,
e.field, e.old, e.new))
def history(self, card_id: str) -> list:
rows = self._sql(f"SELECT {_EVENT} FROM activity WHERE card_id = %s ORDER BY id",
(card_id,))
return [_event(r) for r in rows]
def activity(self, board_id: str, limit: int) -> list:
rows = self._sql(f"SELECT {_EVENT} FROM activity WHERE board_id = %s "
"ORDER BY id DESC LIMIT %s", (board_id, limit))
return [_event(r) for r in rows]
def embeddings(self, board_id: str) -> dict:
rows = self._sql("SELECT card_id, text_hash, embedding FROM card_embeddings "
"WHERE board_id = %s", (board_id,))
return {card_id: (digest, json.loads(vector)) for card_id, digest, vector in rows}
def save_embedding(self, board_id: str, card_id: str, text_hash: str, vector: list) -> None:
self._sql("INSERT INTO card_embeddings (card_id, board_id, text_hash, embedding) "
"VALUES (%s, %s, %s, %s) ON CONFLICT (card_id) DO UPDATE SET "
"text_hash = EXCLUDED.text_hash, embedding = EXCLUDED.embedding",
(card_id, board_id, text_hash, json.dumps(vector)))