rgboards/rg-boards
Lanes, team task boards you run from the web or from Claude/ChatGPT (MCP). FastAPI + DataK3 + appid + Dodil Models.
A remix template · 0 remixes · Remix rgboards/rg-boards
scripts/seed_demo.py
"""Seed three realistic demo boards through the same use cases as the web page and MCP, so every
card has a real number, history, labels, assignees, due dates, blocked-by links and checklists.
cd app
# against your bucket (your own dodil token, or the service account env vars):
BUCKET=<repo>-data DODIL_TOKEN=... python ../scripts/seed_demo.py [--reset]
[--owner a@team.com --member b@team.com --viewer c@team.com]
Due dates are relative to today, so the demo always has a few overdue cards, a few due this
week and the rest later. The history is spread over the last working days (office hours, a
few minutes to a few hours apart), so it reads like a team's week instead of one instant.
--reset first deletes the boards these three people own.
Who ends up where (the point of the demo is the roles):
Product Launch owner=OWNER, members=MEMBER + 2 teammates, viewer=VIEWER (16 cards)
Ops owner=OWNER, members=MEMBER + 1 teammate; VIEWER is NOT on it
Website Refresh owner=MEMBER, member=OWNER + 1 teammate, viewer=VIEWER
"""
import argparse
import sys
from datetime import date, datetime, time, timedelta, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "app")) # run from anywhere
from boards.application.use_cases import Boards # noqa: E402
from boards.domain.model import Person # noqa: E402
TABLES = ("cards", "memberships", "activity", "card_links", "checklist_items", "card_vectors",
"board_summaries")
def reset(repo, emails) -> None:
run = getattr(repo, "_run", None)
if run is None:
return # in-memory store: nothing to clear
for email in emails:
for (bid,) in run("SELECT id FROM boards WHERE owner_email = %s", (email,)):
for t in TABLES:
run(f"DELETE FROM {t} WHERE board_id = %s", (bid,), False)
run("DELETE FROM boards WHERE id = %s", (bid,), False)
class WorkWeekClock:
"""A clock for the seed's history: starts `days` working days ago at 09:30 IST and moves a
few minutes to a few hours per event, skipping nights and weekends, never past `now`."""
STEPS = (34, 7, 82, 15, 128, 22, 51, 9, 167, 41, 12, 96) # minutes, a deterministic jitter
def __init__(self, now: datetime, days: int = 6):
self.now, self.i = now, 0
start = now.date()
while days:
start -= timedelta(days=1)
days -= start.weekday() < 5
self.t = datetime.combine(start, time(4, 0), timezone.utc) # 09:30 IST
def __call__(self) -> datetime:
self.t += timedelta(minutes=self.STEPS[self.i % len(self.STEPS)])
self.i += 1
if self.t.hour >= 13: # after ~18:30 IST: next working morning
self.t = datetime.combine(self.t.date() + timedelta(days=1), time(4, 0),
timezone.utc) + timedelta(minutes=self.i % 25)
while self.t.weekday() >= 5:
self.t += timedelta(days=1)
return min(self.t, self.now - timedelta(minutes=5))
def seed(s: Boards, owner: str, member: str, viewer: str, today: date = None,
now: datetime = None) -> dict:
real_clock = s.now
s.now = WorkWeekClock(now or datetime.now(timezone.utc))
try:
return _seed(s, owner, member, viewer, today)
finally:
s.now = real_clock
def _seed(s: Boards, owner: str, member: str, viewer: str, today: date = None) -> dict:
today = today or date.today()
d = lambda n: (today + timedelta(days=n)).isoformat() # noqa: E731
O, M = Person(owner), Person(member)
dom = owner.split("@")[1]
priya, marco, aisha = (f"{n}@{dom}" for n in ("priya.shah", "marco.rossi", "aisha.khan"))
# ---------------------------------------------------------------- Product Launch
pl = s.create_board(O, "Product Launch")["id"]
for email, role in ((member, "member"), (priya, "member"), (marco, "member"),
(viewer, "viewer")):
s.add_member(O, pl, email, role)
s.add_list(O, pl, "Review")
s.add_list(O, pl, "Backlog") # lands before Done; moved to the front below
def card(who, title, lst, assignee=None, due=None, labels=(), desc=""): # noqa: E306
return s.add_card(who, pl, title, list_name=lst, description=desc, assignee=assignee,
due_date=due, labels=list(labels))
c1 = card(O, "Write launch blog post", "To do", priya, d(-2), ["content"],
"800 words: the problem, the product story, 3 screenshots, one customer quote.")
c2 = card(O, "Final QA pass on checkout", "To do", marco, d(7), ["qa"],
"Every checkout path on staging: cards, coupons, refunds, receipts.")
c3 = card(O, "Ship v1 to production", "To do", owner, d(13), ["release"],
"Tag v1.0, run the migration, flip the feature flag, watch errors for an hour.")
c4 = card(O, "Record 60-second demo video", "To do", member, d(11), ["content", "video"])
c5 = card(O, "Pricing page copy", "To do", priya, None, ["content"])
c6 = card(O, "Set up billing sandbox", "To do", marco, None, ["billing"])
c7 = card(O, "Draft press email to 20 journalists", "To do", owner, d(-1), ["pr"],
"Short pitch, embargo until launch morning, link to the press kit.")
c8 = card(O, "Launch day announcement: blog, social, email", "To do", member, d(21),
["marketing"])
c9 = card(O, "Fix login redirect bug on mobile Safari", "To do", marco, d(-3), ["bug"],
"After sign-in, iOS Safari lands on /undefined. Repro: private tab, sign in.")
c10 = card(O, "Security review of auth and payment flows", "To do", member, d(6),
["security"])
c11 = card(O, "Customer onboarding email sequence", "To do", priya, d(26),
["content", "email"])
card(O, "Update help center for v1", "To do", member, d(19), ["docs"])
c13 = card(O, "Load test checkout at 10x traffic", "To do", marco, d(5), ["qa", "perf"])
c14 = card(O, "Collect beta testimonials", "To do", owner, None, ["marketing"])
c15 = card(O, "Pick the launch date with sales", "To do", owner, None, [])
c16 = card(O, "Post-launch retro", "To do", owner, d(35), ["team"])
# the work happening: moves and edits by different people, so history reads like a team
for c, lst, who in ((c5, "Doing", M), (c6, "Doing", M), (c14, "Doing", O),
(c5, "Done", M), (c6, "Done", M), (c14, "Done", O), (c15, "Done", O),
(c1, "Doing", M), (c2, "Doing", M), (c13, "Doing", M),
(c7, "Review", O), (c10, "Review", M),
(c8, "Backlog", O), (c11, "Backlog", O), (c16, "Backlog", O)):
s.move_card(who, c["id"], lst)
s.update_card(M, c2["id"], description="Every checkout path on staging: cards, coupons, "
"refunds, receipts. Blocked until the mobile login fix lands.")
s.update_card(O, c3["id"], due_date=d(14))
# the blocked-by chain: launch day <- ship v1 <- QA <- mobile login bug; ship v1 <- security
s.add_blocker(O, c8["id"], c3["number"])
s.add_blocker(O, c3["id"], c2["number"])
s.add_blocker(O, c3["id"], c10["number"])
s.add_blocker(M, c2["id"], c9["number"])
s.add_blocker(M, c1["id"], c4["number"])
qa = s.make_checklist(M, c2["id"], items=["Card payments", "Coupon codes", "Refunds",
"Email receipts", "Mobile Safari"])
for item in qa["checklist"][:2]:
s.tick_item(M, c2["id"], item["id"], True)
s.make_checklist(O, c3["id"], items=["Tag v1.0", "Run the migration",
"Flip the feature flag", "Watch errors for an hour"])
# put Backlog first: To do, Doing, Review, Done stay in their order behind it
board = s.repo.get(pl)
board.lists = ["Backlog"] + [n for n in board.lists if n != "Backlog"]
s.repo.save_board(board)
# ---------------------------------------------------------------- Ops (viewer not on it)
ops = s.create_board(O, "Ops")["id"]
s.add_member(O, ops, member, "member")
s.add_member(O, ops, aisha, "member")
def op(title, lst, assignee=None, due=None, labels=(), desc=""): # noqa: E306
return s.add_card(O, ops, title, list_name=lst, description=desc, assignee=assignee,
due_date=due, labels=list(labels))
o1 = op("Send invoice to Acme", "Doing", owner, d(1), ["finance"], "Net 30, PO #4471.")
op("Renew domain names", "To do", member, d(16), ["admin"])
o3 = op("Pay contractor invoices", "To do", aisha, d(-4), ["finance"])
op("Quarterly tax filing", "To do", owner, d(31), ["finance", "compliance"])
op("Rotate API keys", "Done", member, None, ["security"])
op("Order laptops for two new hires", "Doing", aisha, d(9), ["admin"])
o7 = op("Close October books", "To do", owner, d(8), ["finance"])
s.add_blocker(O, o7["id"], o1["number"])
s.add_blocker(O, o7["id"], o3["number"])
# ---------------------------------------------------------------- Website Refresh
wr = s.create_board(M, "Website Refresh")["id"] # the member owns this one
s.add_member(M, wr, owner, "member")
s.add_member(M, wr, priya, "member")
s.add_member(M, wr, viewer, "viewer")
def w(title, lst, assignee=None, due=None, labels=()): # noqa: E306
return s.add_card(M, wr, title, list_name=lst, assignee=assignee, due_date=due,
labels=list(labels))
w1 = w("New homepage hero copy", "Doing", priya, d(7), ["content", "design"])
w("Pick the new colour palette", "Done", member, None, ["design"])
w3 = w("Rebuild the pricing table component", "To do", owner, d(14), ["frontend"])
w("Compress hero images under 200 KB", "To do", member, d(-1), ["perf"])
w("Accessibility audit (WCAG AA)", "To do", priya, d(22), ["a11y"])
w("Redirects for old blog URLs", "Doing", member, d(4), ["seo"])
s.add_blocker(M, w3["id"], w1["number"])
return {"Product Launch": pl, "Ops": ops, "Website Refresh": wr}
def main() -> None:
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
ap.add_argument("--owner", default="demo.owner@lanes-demo.dev")
ap.add_argument("--member", default="demo.member@lanes-demo.dev")
ap.add_argument("--viewer", default="demo.viewer@lanes-demo.dev")
ap.add_argument("--reset", action="store_true", help="delete their boards first")
a = ap.parse_args()
from boards.infrastructure.repository import SqlBoards
repo = SqlBoards()
if a.reset:
reset(repo, [a.owner, a.member])
print(seed(Boards(repo, None), a.owner.lower(), a.member.lower(), a.viewer.lower()))
if __name__ == "__main__":
main()