mobiledev/taskflow
TaskFlow, a team task board you run from the web or from Claude and ChatGPT. FastAPI + DataK3 + appid + Models + MCP.
A remix template · 0 remixes · Remix mobiledev/taskflow
app/taskflow/infrastructure/repository.py
"""Boards and cards stored in DataK3 tables (schema.sql). Writes are keyed upserts, so a retry is
safe. Lists, labels, dependencies and checklists are small JSON documents in TEXT columns."""
import json
import uuid
from typing import Optional
from psycopg import errors
from taskflow.domain.model import Board, Card, Role
from taskflow.infrastructure.dodil import connect
CARD_COLUMNS = ("id, board_id, number, title, list_name, position, description, assignee, "
"due_date, labels, blocked_by, checklist")
def _card(row) -> Card:
(id_, board_id, number, title, list_name, position, description, assignee, due_date,
labels, blocked_by, checklist) = row
return Card(id=id_, board_id=board_id, number=int(number), title=title,
list_name=list_name, position=float(position or 0), description=description or "",
assignee=assignee, due_date=due_date, labels=tuple(json.loads(labels or "[]")),
blocked_by=json.loads(blocked_by or "[]"), checklist=json.loads(checklist or "[]"))
class SqlBoards:
def get(self, board_id: str) -> Optional[Board]:
with connect() as c:
row = c.execute("SELECT id, title, owner_email, lists, card_seq 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], lists=json.loads(row[3]),
card_seq=int(row[4] or 0), members={e: Role(r) for e, r in members})
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))]
return [b for b in (self.get(i) for i in ids) if b]
def save_board(self, board: Board) -> None:
with connect() as c:
c.execute("INSERT INTO boards (id, title, owner_email, lists, card_seq) "
"VALUES (%s, %s, %s, %s, %s) ON CONFLICT (id) DO UPDATE SET "
"title = EXCLUDED.title, lists = EXCLUDED.lists, "
"card_seq = EXCLUDED.card_seq",
(board.id, board.title, board.owner_email, json.dumps(board.lists),
board.card_seq))
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}",))
def get_card(self, card_id: str) -> Optional[Card]:
with connect() as c:
row = c.execute(f"SELECT {CARD_COLUMNS} FROM cards WHERE id = %s",
(card_id,)).fetchone()
return _card(row) if row else None
def cards_on(self, board_id: str) -> list:
with connect() as c:
rows = c.execute(f"SELECT {CARD_COLUMNS} FROM cards WHERE board_id = %s "
"ORDER BY position", (board_id,)).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_COLUMNS}, updated_at) VALUES "
"(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, now()) "
"ON CONFLICT (id) DO UPDATE SET title = EXCLUDED.title, "
"list_name = EXCLUDED.list_name, position = EXCLUDED.position, "
"description = EXCLUDED.description, assignee = EXCLUDED.assignee, "
"due_date = EXCLUDED.due_date, labels = EXCLUDED.labels, "
"blocked_by = EXCLUDED.blocked_by, checklist = EXCLUDED.checklist, "
"updated_at = EXCLUDED.updated_at",
(card.id, card.board_id, card.number, card.title, card.list_name,
card.position, card.description, card.assignee, card.due_date,
json.dumps(list(card.labels)), json.dumps(card.blocked_by),
json.dumps(card.checklist)))
def delete_card(self, card_id: str) -> None:
with connect() as c:
c.execute("DELETE FROM cards WHERE id = %s", (card_id,))
def delete_board(self, board_id: str) -> None:
with connect() as c:
for table in ("cards", "card_numbers", "memberships", "activity"):
c.execute(f"DELETE FROM {table} WHERE board_id = %s", (board_id,))
c.execute("DELETE FROM boards WHERE id = %s", (board_id,))
def claim_card_number(self, board_id: str, at_least: int) -> int:
"""A claim is a plain INSERT: under concurrency exactly one writer gets each number,
the others see a unique violation and try the next one. (An upsert would let both win.)"""
with connect() as c:
for n in range(at_least, at_least + 100):
try:
c.execute("INSERT INTO card_numbers (id, board_id, number) "
"VALUES (%s, %s, %s)", (f"{board_id}:{n}", board_id, n))
return n
except errors.UniqueViolation:
continue
raise RuntimeError("could not claim a card number")
def table_counts(self) -> dict:
"""For the deploy self-check: proves the app reaches its own bucket."""
with connect() as c:
return {t: c.execute(f"SELECT count(*) FROM {t}").fetchone()[0]
for t in ("boards", "memberships", "cards", "card_numbers", "activity")}
def log(self, board_id: str, card_id: Optional[str], actor: str, action: str,
changes: Optional[dict] = None) -> None:
with connect() as c:
c.execute("INSERT INTO activity (id, board_id, card_id, actor, action, changes, "
"event_at) VALUES (%s, %s, %s, %s, %s, %s, now())",
(uuid.uuid4().hex, board_id, card_id, actor, action,
json.dumps(changes or {})))
def activity(self, board_id: str, card_id: Optional[str] = None, limit: int = 50) -> list:
sql = ("SELECT event_at, actor, action, changes FROM activity WHERE board_id = %s"
+ (" AND card_id = %s" if card_id else "") + " ORDER BY event_at DESC LIMIT %s")
args = (board_id, card_id, limit) if card_id else (board_id, limit)
with connect() as c:
rows = c.execute(sql, args).fetchall()
return [{"at": at.isoformat() if hasattr(at, "isoformat") else str(at), "actor": actor,
"action": action, "changes": json.loads(changes or "{}")}
for at, actor, action, changes in rows]