freelancer/flowboard
Boards starter for the Dodil Boards Challenge. FastAPI + DataK3 + appid + Models + MCP.
A remix template · 0 remixes · Remix freelancer/flowboard
app/boards/infrastructure/repository.py
"""DataK3 repository for boards, cards, memberships and activity."""
import json
import threading
import uuid
from typing import Optional
from psycopg.types.json import Jsonb
from boards.domain.model import Board, Card, Role
from boards.infrastructure.dodil import connect
class SqlBoards:
"""Persistent repository with backward-compatible card migration."""
def __init__(self):
self._schema_ready = False
self._schema_lock = threading.Lock()
def _ensure_card_schema(self) -> None:
"""Add card fields without deleting existing data."""
if self._schema_ready:
return
with self._schema_lock:
if self._schema_ready:
return
with connect() as connection:
connection.execute(
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS "
"description TEXT NOT NULL DEFAULT ''"
)
connection.execute(
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS labels JSONB"
)
self._schema_ready = True
@staticmethod
def _card(row) -> Card:
"""Convert a database row into the domain model."""
labels = row[7] if row[7] is not None else []
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=list(labels),
)
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:
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",
(email, email),
)
]
return [
board
for board in (self.get(board_id) for board_id in ids)
if board
]
def save_board(self, board: Board) -> None:
with connect() as connection:
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),
)
for email, role in board.members.items():
connection.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 get_card(self, card_id: str) -> Optional[Card]:
self._ensure_card_schema()
with connect() as connection:
row = connection.execute(
"SELECT id, board_id, title, list_name, assignee, due_date, "
"description, labels FROM cards WHERE id = %s",
(card_id,),
).fetchone()
return self._card(row) if row else None
def cards_on(self, board_id: str) -> list:
self._ensure_card_schema()
with connect() as connection:
rows = connection.execute(
"SELECT id, board_id, title, list_name, assignee, due_date, "
"description, labels FROM cards "
"WHERE board_id = %s ORDER BY updated_at",
(board_id,),
).fetchall()
return [self._card(row) for row in rows]
def save_card(self, card: Card) -> None:
self._ensure_card_schema()
with connect() as connection:
connection.execute(
"INSERT INTO cards "
"(id, board_id, list_name, title, assignee, due_date, "
"description, labels, updated_at) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, now()) "
"ON CONFLICT (id) DO UPDATE SET "
"list_name = EXCLUDED.list_name, "
"title = EXCLUDED.title, "
"assignee = EXCLUDED.assignee, "
"due_date = EXCLUDED.due_date, "
"description = EXCLUDED.description, "
"labels = EXCLUDED.labels, "
"updated_at = EXCLUDED.updated_at",
(
card.id,
card.board_id,
card.list_name,
card.title,
card.assignee,
card.due_date,
card.description,
Jsonb(card.labels),
),
)
def log(
self,
board_id: str,
card_id: Optional[str],
actor: str,
action: str,
) -> None:
with connect() as connection:
connection.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 activity_for(
self,
board_id: str,
limit: int = 100,
) -> list[dict]:
"""Return recent activity for a board, newest first."""
safe_limit = max(1, min(int(limit), 500))
with connect() as connection:
rows = connection.execute(
"SELECT id, board_id, card_id, actor, action, event_at "
"FROM activity "
"WHERE board_id = %s "
"ORDER BY event_at DESC, id DESC "
"LIMIT %s",
(board_id, safe_limit),
).fetchall()
return [
{
"id": row[0],
"board_id": row[1],
"card_id": row[2],
"actor": row[3],
"action": row[4],
"event_at": (
row[5].isoformat() if row[5] is not None else None
),
}
for row in rows
]