eduard/cobra-boards
Cobra Boards, team task boards you can run from Claude or ChatGPT. FastAPI + DataK3 + appid + Dodil Models + MCP.
A remix template · 1 remixes · Remix eduard/cobra-boards
scripts/seed_demo_team.py
"""Seeds the shared demo team on a live copy, for judges: one board with an owner, a member and a
viewer (the accounts are created by the org admin in the app's pool; this script never touches
passwords).
It builds the boards through the normal use cases, in memory, so every row obeys the app's rules,
then writes those rows into the bucket with the `dodil` CLI (`data sql` and `data object create`),
signed in as the person running it. Only INSERTs: it never changes or removes existing data.
python scripts/seed_demo_team.py <bucket> owner@x member@y viewer@z [--dry-run]
"""
import json
import os
import subprocess
import sys
import tempfile
from datetime import date, datetime
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app"))
sys.path.insert(0, os.path.dirname(__file__))
import preview # noqa: E402 (the demo story, shared with the local preview)
from boards.application.use_cases import Boards # noqa: E402
from boards.infrastructure.memory import MemoryBoards, MemoryObjects # noqa: E402
DODIL = os.environ.get("DODIL", os.path.expanduser("~/.local/bin/dodil.exe"))
def lit(v) -> str:
"""A SQL literal for one value (the rows come from our own use cases, not from users)."""
if v is None:
return "NULL"
if isinstance(v, bool):
return "TRUE" if v else "FALSE"
if isinstance(v, (int, float)):
return repr(v)
if isinstance(v, datetime):
return "'" + v.strftime("%Y-%m-%d %H:%M:%S") + "'"
if isinstance(v, date):
return "'" + v.isoformat() + "'"
return "'" + str(v).replace("'", "''") + "'"
def insert(table: str, row: dict) -> str:
return (f"INSERT INTO {table} ({', '.join(row)}) VALUES "
f"({', '.join(lit(v) for v in row.values())})")
def rows(repo: MemoryBoards) -> list:
out = []
for b in repo.boards.values():
out.append(insert("boards", {"id": b.id, "title": b.title, "owner_email": b.owner_email,
"created_at": b.created_at}))
for email, role in b.members.items():
out.append(insert("memberships", {"id": f"{b.id}:{email}", "board_id": b.id,
"email": email, "role": role.value}))
for lst in b.lists:
out.append(insert("board_lists", {"id": lst.id, "board_id": b.id, "name": lst.name,
"position": lst.position, "is_done": lst.is_done}))
for c in repo.cards.values():
out.append(insert("cards", {
"id": c.id, "board_id": c.board_id, "number": c.number, "list_id": c.list_id,
"title": c.title, "description": c.description, "assignee": c.assignee,
"due_date": c.due_date, "labels": json.dumps(c.labels), "position": c.position,
"version": c.version, "created_by": c.created_by, "created_at": c.created_at,
"updated_at": c.updated_at}))
for e in repo.events:
out.append(insert("card_events", {
"id": e.id, "board_id": e.board_id, "card_id": e.card_id, "actor": e.actor,
"action": e.action, "summary": e.summary, "changes": json.dumps(e.changes, default=str),
"event_at": e.at}))
for board_id, card_id, blocker_id in repo.deps:
out.append(insert("card_dependencies", {"id": f"{card_id}<{blocker_id}",
"board_id": board_id, "card_id": card_id,
"blocker_id": blocker_id}))
for card_id, items in repo.checklists.items():
for it in items.values():
out.append(insert("checklist_items", {"id": it["id"], "card_id": card_id,
"text": it["text"], "done": it["done"],
"position": it["position"]}))
for cm in repo.comment_rows:
out.append(insert("card_comments", {**cm, "mentions": json.dumps(cm["mentions"])}))
for n in repo.notes:
out.append(insert("notifications", {k: n.get(k) for k in (
"id", "email", "board_id", "card_id", "kind", "actor", "summary", "created_at",
"read_at")}))
for a in repo.atts.values():
out.append(insert("card_attachments", a))
for card_id, info in repo.archived.items():
out.append(insert("archived_cards", {"card_id": card_id, "board_id": info["board_id"],
"archived_by": info["by"], "archived_at": info["at"]}))
return out
def main():
args = [a for a in sys.argv[1:] if not a.startswith("--")]
dry = "--dry-run" in sys.argv
bucket, owner, member, viewer = args
preview.TEAM.update(owner=owner, member=member, viewer=viewer)
s = Boards(MemoryBoards(), preview.DemoAI(), objects=MemoryObjects())
preview.seed(s)
statements = rows(s.repo)
print(f"{len(s.repo.boards)} boards, {len(s.repo.cards)} cards, {len(statements)} rows, "
f"{len(s.objects.blobs)} files -> bucket {bucket}" + (" (dry run)" if dry else ""))
if dry:
return
for key, (data, _ctype) in s.objects.blobs.items():
with tempfile.NamedTemporaryFile(delete=False) as f:
f.write(data)
r = subprocess.run([DODIL, "data", "object", "create", "-b", bucket, f.name, "-k", key],
capture_output=True, text=True, encoding="utf-8", errors="replace")
os.unlink(f.name)
print("object", key, "ok" if r.returncode == 0 else "FAILED " + r.stderr[-200:])
failed = 0
for i, st in enumerate(statements, 1):
r = subprocess.run([DODIL, "data", "sql", "-b", bucket, st], capture_output=True,
text=True, encoding="utf-8", errors="replace")
if r.returncode != 0:
failed += 1
print("FAILED", st[:90], (r.stderr or r.stdout)[-200:])
if i % 20 == 0:
print(f" {i}/{len(statements)}")
print("done," if not failed else f"{failed} failed,", f"{len(statements)} rows")
if __name__ == "__main__":
main()