contest/orbit-boards
Orbit Boards, a private, remixable team task board built with FastAPI, DataK3, AppID, Models and MCP.
A remix template · 0 remixes · Remix contest/orbit-boards
app/boards/infrastructure/repository.py
"""DataK3 persistence. Mutations and their audit records commit in one transaction."""
import json
import uuid
from typing import Optional
from boards.domain.model import Board, Card, Role
from boards.infrastructure.dodil import connect
_CARD_FIELDS = "id, board_id, title, list_name, assignee, due_date, description, labels, position"
def _card(row) -> Card:
labels = row[7] or "[]"
if isinstance(labels, str):
labels = json.loads(labels)
return Card(id=row[0], board_id=row[1], title=row[2], list_name=row[3],
assignee=row[4], due_date=row[5], description=row[6] or "",
labels=labels, position=row[8])
class SqlBoards:
def check_ready(self) -> None:
"""Verify connectivity and the columns required by the deployed repository."""
with connect() as connection:
connection.execute("SELECT id, title, owner_email, created_at FROM boards LIMIT 0")
connection.execute("SELECT id, board_id, email, role FROM memberships LIMIT 0")
connection.execute(
"SELECT id, board_id, title, list_name, assignee, due_date, description, "
"labels, position, created_at, updated_at FROM cards LIMIT 0")
connection.execute(
"SELECT id, board_id, card_id, actor, action, field, old_value, new_value, "
"event_at FROM activity LIMIT 0")
def get(self, board_id: str) -> Optional[Board]:
with connect() as connection:
row = connection.execute("SELECT id, title, owner_email FROM boards WHERE id = %s",
(board_id,)).fetchone()
if not row:
return None
members = connection.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={email: Role(role) for email, role in members})
def boards_for(self, email: str) -> list[Board]:
with connect() as connection:
ids = [row[0] for row in connection.execute(
"SELECT id FROM boards WHERE owner_email = %s "
"UNION SELECT board_id FROM memberships WHERE email = %s ORDER BY id",
(email.strip().lower(), email.strip().lower()))]
return [board for board in (self.get(board_id) for board_id in ids) if board]
@staticmethod
def _save_board(connection, board: Board) -> None:
connection.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))
connection.execute("DELETE FROM memberships WHERE board_id = %s", (board.id,))
for email, role in board.members.items():
connection.execute(
"INSERT INTO memberships (id, board_id, email, role) VALUES (%s, %s, %s, %s)",
(f"{board.id}:{email}", board.id, email, role.value))
@staticmethod
def _save_event(connection, event: dict) -> None:
connection.execute(
"INSERT INTO activity (id, board_id, card_id, actor, action, field, old_value, "
"new_value, event_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, now())",
(uuid.uuid4().hex, event["board_id"], event.get("card_id"), event["actor"],
event["action"], event.get("field"),
json.dumps(event.get("old_value"), ensure_ascii=False),
json.dumps(event.get("new_value"), ensure_ascii=False)))
def save_board(self, board: Board) -> None:
with connect() as connection:
self._save_board(connection, board)
def save_board_event(self, board: Board, event: dict) -> None:
with connect() as connection:
self._save_board(connection, board)
self._save_event(connection, event)
def get_card(self, card_id: str) -> Optional[Card]:
with connect() as connection:
row = connection.execute(f"SELECT {_CARD_FIELDS} FROM cards WHERE id = %s",
(card_id,)).fetchone()
return _card(row) if row else None
def cards_on(self, board_id: str) -> list[Card]:
with connect() as connection:
rows = connection.execute(
f"SELECT {_CARD_FIELDS} FROM cards WHERE board_id = %s "
"ORDER BY list_name, position, id", (board_id,)).fetchall()
return [_card(row) for row in rows]
@staticmethod
def _save_card(connection, card: Card) -> None:
connection.execute(
"INSERT INTO cards (id, board_id, title, list_name, assignee, due_date, description, "
"labels, position, updated_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, now()) "
"ON CONFLICT (id) DO UPDATE SET title = EXCLUDED.title, list_name = EXCLUDED.list_name, "
"assignee = EXCLUDED.assignee, due_date = EXCLUDED.due_date, "
"description = EXCLUDED.description, labels = EXCLUDED.labels, "
"position = EXCLUDED.position, updated_at = EXCLUDED.updated_at",
(card.id, card.board_id, card.title, card.list_name, card.assignee, card.due_date,
card.description, json.dumps(card.labels, ensure_ascii=False), card.position))
def save_card(self, card: Card) -> None:
with connect() as connection:
self._save_card(connection, card)
def save_cards_and_events(self, cards: list[Card], events: list[dict]) -> None:
with connect() as connection:
for card in cards:
self._save_card(connection, card)
for event in events:
self._save_event(connection, event)
def log(self, board_id: str, card_id: Optional[str], actor: str, action: str) -> None:
with connect() as connection:
self._save_event(connection, {"board_id": board_id, "card_id": card_id,
"actor": actor, "action": action})
def activity_for(self, board_id: str, limit: int = 50) -> list[dict]:
with connect() as connection:
rows = connection.execute(
"SELECT id, board_id, card_id, actor, action, field, old_value, new_value, event_at "
"FROM activity WHERE board_id = %s ORDER BY event_at DESC, id DESC LIMIT %s",
(board_id, limit)).fetchall()
return [{"id": row[0], "board_id": row[1], "card_id": row[2], "actor": row[3],
"action": row[4], "field": row[5],
"old_value": json.loads(row[6]) if row[6] is not None else None,
"new_value": json.loads(row[7]) if row[7] is not None else None,
"event_at": row[8].isoformat() if row[8] else None} for row in rows]