feat(palier-3): recherche globale, mes issues, briefing ntfy, MCP lecture

- client: search_issues (/repos/issues/search, type=issues,
  assigned_by), delete_label + forget_label_cache; models: kv_store
- /search groupe par projet avec colonne Kanban; /my = issues
  ouvertes assignées à l'utilisateur du jeton (get_user)
- briefing: _briefing_data (focus, âgées, milestones <= 7 j,
  activité journal), page /briefing, /api/briefing, envoi ntfy
  manuel + planificateur lifespan à KANBANTE_BRIEFING_TIME
  (kv briefing_last_sent, pas de doublon); app/briefing.py =
  format texte + envoi ntfy (en-têtes ASCII)
- MCP: POST /mcp JSON-RPC (initialize, tools/list, tools/call,
  ping, notifications 202), 7 outils en lecture seule, GET /mcp
  de découverte; erreurs -32601/-32700, isError pour les outils
- focus refactoré en _collect_focus_groups réutilisé par le
  briefing et l'outil MCP get_focus
This commit is contained in:
Spark
2026-10-11 11:00:53 -04:00
parent 2d89db8213
commit a788db7b1f
5 changed files with 718 additions and 6 deletions
+71
View File
@@ -0,0 +1,71 @@
"""Daily briefing: formatting and ntfy delivery.
The data itself is gathered in app.main (it needs the board helpers);
this module only turns it into a readable text and pushes it to an
ntfy topic. Kept dependency-light so it is unit-testable alone.
"""
from __future__ import annotations
import httpx
from app.config import Settings
def ntfy_configured(settings: Settings) -> bool:
return bool(settings.ntfy_url and settings.ntfy_topic)
def format_briefing_text(data: dict) -> str:
"""Plain-text briefing (ntfy body, also shown on /briefing)."""
lines: list[str] = [f"📋 Kanbante — briefing du {data['date']}"]
activity = data["activity_24h"]
lines.append(
f"\n⚡ Dernières 24 h : {activity['closed']} fermée(s), "
f"{activity['created']} créée(s), {activity['moved']} déplacement(s)."
)
if activity.get("avg_cycle_days") is not None:
lines.append(f"⏱️ Temps de cycle moyen (30 j) : {activity['avg_cycle_days']} j.")
lines.append(f"\n🎯 En cours / en révision : {data['focus_total']} carte(s)")
for group in data["focus_groups"][:6]:
lines.append(f" • {group['full_name']} : {group['count']}")
for title in group["titles"][:3]:
lines.append(f" – {title}")
if data["aging"]:
lines.append("\n⏳ Cartes qui vieillissent :")
for card in data["aging"][:8]:
lines.append(
f" • {card['repo']} #{card['number']} « {card['title']} » "
f"— {card['days']} j en {card['column_label']}"
)
if data["milestones_due"]:
lines.append("\n📅 Milestones proches :")
for ms in data["milestones_due"][:5]:
lines.append(
f" • {ms['repo']} « {ms['title']} » — échéance "
f"{ms['due_on']} ({ms['open']} ouverte(s))"
)
if data["focus_total"] == 0 and not data["aging"]:
lines.append("\n🎉 Rien d'urgent : aucun encours, aucune carte vieillissante.")
return "\n".join(lines)
async def send_ntfy(settings: Settings, text: str) -> bool:
"""POST the briefing to the configured ntfy topic. True on 2xx."""
if not ntfy_configured(settings):
return False
url = f"{settings.ntfy_url.rstrip('/')}/{settings.ntfy_topic}"
headers = {
# ntfy headers must stay ASCII; the body carries the accents.
"Title": "Kanbante - briefing du jour",
"Tags": "clipboard,chart_with_upwards_trend",
}
if settings.ntfy_token:
headers["Authorization"] = f"Bearer {settings.ntfy_token}"
async with httpx.AsyncClient(timeout=15.0) as client:
response = await client.post(url, content=text.encode("utf-8"), headers=headers)
return response.status_code < 300
+17 -1
View File
@@ -28,6 +28,14 @@ Environment variables:
X-Gitea-Signature HMAC is enforced.
KANBANTE_SYNC_LABELS "1" (default) = mirror intermediate
columns to Gitea labels named kanban:<col>.
KANBANTE_NTFY_URL Base URL of an ntfy server (e.g.
https://ntfy.sh or self-hosted).
KANBANTE_NTFY_TOPIC ntfy topic for the daily briefing.
KANBANTE_NTFY_TOKEN Optional Bearer token for ntfy.
KANBANTE_BRIEFING_TIME "HH:MM" (server local time) at which
the daily briefing is sent to ntfy.
Empty (default) = no automatic send,
the /briefing page stays available.
"""
import os
@@ -36,7 +44,7 @@ from pathlib import Path
BASE_DIR = Path(__file__).resolve().parent.parent
APP_VERSION = "0.4.0"
APP_VERSION = "0.5.0"
def _env_bool(name: str, default: bool) -> bool:
@@ -88,6 +96,10 @@ class Settings:
aging_days: int = 7
webhook_secret: str = ""
sync_labels: bool = True
ntfy_url: str = ""
ntfy_topic: str = ""
ntfy_token: str = ""
briefing_time: str = ""
@property
def auth_enabled(self) -> bool:
@@ -113,4 +125,8 @@ def get_settings() -> Settings:
aging_days=_env_int("KANBANTE_AGING_DAYS", 7),
webhook_secret=os.getenv("KANBANTE_WEBHOOK_SECRET", ""),
sync_labels=_env_bool("KANBANTE_SYNC_LABELS", True),
ntfy_url=os.getenv("KANBANTE_NTFY_URL", ""),
ntfy_topic=os.getenv("KANBANTE_NTFY_TOPIC", ""),
ntfy_token=os.getenv("KANBANTE_NTFY_TOKEN", ""),
briefing_time=os.getenv("KANBANTE_BRIEFING_TIME", ""),
)
+31
View File
@@ -269,6 +269,37 @@ class GiteaClient:
)
r.raise_for_status()
async def delete_label(self, owner: str, repo: str, label_id: int) -> None:
"""Delete a label definition from the repo (all issues lose it)."""
r = await self.client.delete(f"/repos/{owner}/{repo}/labels/{label_id}")
r.raise_for_status()
self.forget_label_cache(owner, repo)
def forget_label_cache(self, owner: str, repo: str) -> None:
cache = self.__dict__.get("_label_cache")
if cache:
cache.pop(f"{owner}/{repo}", None)
# ── Issue search (global + assigned) ───────────────────
async def search_issues(
self,
q: str = "",
state: str = "open",
assigned_by: str | None = None,
) -> list[dict]:
"""Search issues across every repo visible to the token.
Uses Gitea's /repos/issues/search (type=issues excludes PRs).
Results carry their repository in issue["repository"].
"""
params: dict[str, Any] = {"state": state, "type": "issues"}
if q:
params["q"] = q
if assigned_by:
params["assigned_by"] = assigned_by
return await self._get_all_pages("/repos/issues/search", **params)
# ── Milestones ─────────────────────────────────────────
async def get_milestones(
+577 -5
View File
@@ -34,11 +34,13 @@ from starlette.middleware.base import BaseHTTPMiddleware
from app.config import APP_VERSION, get_settings
from app.gitea_client import GiteaClient
from app.briefing import format_briefing_text, ntfy_configured, send_ntfy
from app.models import (
KANBAN_COLUMNS,
column_entry_times,
get_board_state,
get_events,
get_kv,
get_metrics,
get_notes,
get_stats,
@@ -48,6 +50,7 @@ from app.models import (
ping_db,
reconcile_board_state,
set_column_order,
set_kv,
set_notes,
)
@@ -86,6 +89,33 @@ EVENT_ICONS = {
# ── App setup ──────────────────────────────────────────────
async def _briefing_scheduler() -> None:
"""Send the daily briefing to ntfy at KANBANTE_BRIEFING_TIME.
Checks once a minute; the kv store remembers the last day sent so
restarts never double-send.
"""
while True:
try:
settings = get_settings()
if settings.briefing_time and ntfy_configured(settings):
now = datetime.now()
today = now.strftime("%Y-%m-%d")
if (
now.strftime("%H:%M") >= settings.briefing_time
and get_kv("briefing_last_sent") != today
):
data = await _briefing_data()
if await send_ntfy(settings, format_briefing_text(data)):
set_kv("briefing_last_sent", today)
logger.info("Daily briefing sent to ntfy (%s)", today)
except asyncio.CancelledError:
raise
except Exception as exc: # never let the scheduler die silently
logger.warning("Briefing scheduler error: %s", exc)
await asyncio.sleep(60)
@asynccontextmanager
async def lifespan(app: FastAPI):
init_db()
@@ -102,7 +132,9 @@ async def lifespan(app: FastAPI):
logger.warning(
"GITEA_TOKEN is not set: only public Gitea content will be visible."
)
scheduler = asyncio.create_task(_briefing_scheduler())
yield
scheduler.cancel()
await gitea.aclose()
@@ -471,9 +503,8 @@ def _card_ages(
return ages
@app.get("/focus", response_class=HTMLResponse)
async def focus(request: Request):
"""Cross-project view: everything in progress or in review."""
async def _collect_focus_groups() -> list[dict]:
"""In-progress / in-review cards across all active repos."""
repos = await gitea.search_repos()
active, _inactive, _archived = classify_repos(repos)
active = active[:100]
@@ -500,15 +531,232 @@ async def focus(request: Request):
]
if cards:
groups.append({"full_name": full_name, "cards": cards})
return groups
@app.get("/focus", response_class=HTMLResponse)
async def focus(request: Request):
"""Cross-project view: everything in progress or in review."""
return render(
request,
"focus.html",
groups=groups,
groups=await _collect_focus_groups(),
column_labels=COLUMN_LABELS,
column_colors=COLUMN_COLORS,
)
def _group_issue_results(issues: list[dict]) -> list[dict]:
"""Group search results by repo, annotated with the board column."""
by_repo: dict[str, list[dict]] = {}
for issue in issues:
full_name = (issue.get("repository") or {}).get("full_name", "?")
by_repo.setdefault(full_name, []).append(issue)
groups = []
for full_name in sorted(by_repo):
state = get_board_state(full_name) if "/" in full_name else {}
column_of = {
card["issue_number"]: column
for column, cards in state.items()
for card in cards
}
rows = []
for issue in by_repo[full_name]:
if issue.get("state") == "open":
column = column_of.get(issue["number"], "backlog")
else:
column = "done"
rows.append({"issue": issue, "column": column})
groups.append({"full_name": full_name, "rows": rows})
return groups
@app.get("/search", response_class=HTMLResponse)
async def search_page(request: Request, q: str = ""):
"""Global issue search across every repo (Gitea search API)."""
q = q.strip()
groups: list[dict] = []
total = 0
if q:
results = await gitea.search_issues(q=q, state="all")
total = len(results)
groups = _group_issue_results(results)
return render(
request,
"search.html",
q=q,
groups=groups,
total=total,
column_labels=COLUMN_LABELS,
column_colors=COLUMN_COLORS,
)
@app.get("/my", response_class=HTMLResponse)
async def my_issues(request: Request):
"""Open issues assigned to the Gitea token's owner, all repos."""
user = None
groups: list[dict] = []
total = 0
try:
user = await gitea.get_user()
except Exception as exc:
logger.debug("get_user failed for /my: %s", exc)
if user and user.get("login"):
results = await gitea.search_issues(
state="open", assigned_by=user["login"]
)
total = len(results)
groups = _group_issue_results(results)
return render(
request,
"my.html",
user=user,
groups=groups,
total=total,
column_labels=COLUMN_LABELS,
column_colors=COLUMN_COLORS,
)
# ── Briefing ───────────────────────────────────────────────
async def _briefing_data() -> dict:
"""Everything the daily briefing needs, gathered in one dict."""
settings = get_settings()
focus_groups = await _collect_focus_groups()
repos = await gitea.search_repos()
active, _i, _a = classify_repos(repos)
soon = datetime.now(timezone.utc) + timedelta(days=7)
aging: list[dict] = []
milestones_due: list[dict] = []
async def _per_repo(repo: dict):
full_name = repo["full_name"]
owner, _, name = full_name.partition("/")
items = await gitea.get_repo_issues(owner, name, state="open")
open_issues = [i for i in items if _is_issue(i)]
ages = _card_ages(full_name, open_issues, get_board_state(full_name))
board = get_board_state(full_name)
column_of = {
card["issue_number"]: column
for column, cards in board.items()
for card in cards
}
repo_aging = [
{
"repo": full_name,
"number": issue["number"],
"title": issue["title"],
"days": ages[issue["number"]]["days"],
"column_label": COLUMN_LABELS[
column_of.get(issue["number"], "backlog")
],
}
for issue in open_issues
if issue["number"] in ages
and ages[issue["number"]]["days"] >= settings.aging_days
]
repo_milestones = []
try:
for ms in await gitea.get_milestones(owner, name):
due = _parse_datetime(ms.get("due_on"))
if due and due <= soon and (ms.get("open_issues") or 0) > 0:
repo_milestones.append(
{
"repo": full_name,
"title": ms.get("title", ""),
"due_on": str(ms["due_on"])[:10],
"open": ms.get("open_issues") or 0,
}
)
except Exception as exc:
logger.debug("briefing milestones failed for %s: %s", full_name, exc)
return repo_aging, repo_milestones
results = await asyncio.gather(
*(_per_repo(r) for r in active[:100]), return_exceptions=True
)
for result in results:
if isinstance(result, Exception):
logger.warning("briefing: repo fetch failed: %s", result)
continue
repo_aging, repo_milestones = result
aging.extend(repo_aging)
milestones_due.extend(repo_milestones)
aging.sort(key=lambda c: -c["days"])
metrics_24h = get_metrics(window_days=1)
metrics_30d = get_metrics(window_days=30)
return {
"date": datetime.now().strftime("%d/%m/%Y"),
"focus_total": sum(len(g["cards"]) for g in focus_groups),
"focus_groups": [
{
"full_name": g["full_name"],
"count": len(g["cards"]),
"titles": [c["issue"]["title"] for c in g["cards"]],
}
for g in focus_groups
],
"aging": aging[:10],
"milestones_due": milestones_due[:5],
"activity_24h": {
**metrics_24h,
"avg_cycle_days": metrics_30d["avg_cycle_days"],
},
}
@app.get("/briefing", response_class=HTMLResponse)
async def briefing_page(request: Request):
"""The daily briefing, readable on the web before it is pushed."""
settings = get_settings()
data = await _briefing_data()
return render(
request,
"briefing.html",
data=data,
text=format_briefing_text(data),
ntfy_configured=ntfy_configured(settings),
briefing_time=settings.briefing_time,
last_sent=get_kv("briefing_last_sent"),
)
@app.get("/api/briefing")
async def api_briefing():
data = await _briefing_data()
return {"ok": True, "briefing": data, "text": format_briefing_text(data)}
@app.post("/api/briefing/send")
async def api_briefing_send(request: Request, csrf_token: str = Form("")):
"""Send the briefing to the configured ntfy topic, right now."""
if not _csrf_valid(request, csrf_token):
return _csrf_error()
settings = get_settings()
if not ntfy_configured(settings):
return JSONResponse(
{
"ok": False,
"error": "ntfy is not configured "
"(KANBANTE_NTFY_URL / KANBANTE_NTFY_TOPIC)",
},
status_code=400,
)
data = await _briefing_data()
sent = await send_ntfy(settings, format_briefing_text(data))
if not sent:
return JSONResponse(
{"ok": False, "error": "ntfy server rejected or unreachable"},
status_code=502,
)
set_kv("briefing_last_sent", datetime.now().strftime("%Y-%m-%d"))
return {"ok": True, "sent": True}
# ── Webhooks + live stream ─────────────────────────────────
def _apply_external_transition(
@@ -802,7 +1050,7 @@ def _issue_payload(issue: dict) -> dict:
"updated_at": issue.get("updated_at", ""),
"due_date": issue.get("due_date"),
"labels": [
{"name": l.get("name", ""), "color": l.get("color", "")}
{"id": l.get("id"), "name": l.get("name", ""), "color": l.get("color", "")}
for l in issue.get("labels") or []
],
"assignees": [
@@ -889,6 +1137,330 @@ async def api_metrics(repo: str = "", window: int = 30):
return {"ok": True, "metrics": get_metrics(repo or None, window)}
# ── Label management ───────────────────────────────────────
@app.get("/api/repos/{owner}/{repo}/labels")
async def api_list_labels(owner: str, repo: str):
"""All label definitions of a repo."""
labels = await gitea.get_labels(owner, repo)
return {
"ok": True,
"labels": [
{"id": l.get("id"), "name": l.get("name", ""), "color": l.get("color", "")}
for l in labels
],
}
@app.post("/api/repos/{owner}/{repo}/labels")
async def api_create_label(
request: Request,
owner: str,
repo: str,
name: str = Form(...),
color: str = Form("7c5cfc"),
csrf_token: str = Form(""),
):
"""Create a label definition in the repo."""
if not _csrf_valid(request, csrf_token):
return _csrf_error()
name = name.strip()
color = color.strip().lstrip("#")
if not name or len(name) > 50:
return JSONResponse(
{"ok": False, "error": "Label name must be 1-50 chars"},
status_code=400,
)
if len(color) != 6 or any(c not in "0123456789abcdefABCDEF" for c in color):
return JSONResponse(
{"ok": False, "error": "Color must be a 6-digit hex value"},
status_code=400,
)
label = await gitea.create_label(owner, repo, name, color)
gitea.forget_label_cache(owner, repo)
return JSONResponse(
{
"ok": True,
"label": {
"id": label.get("id"),
"name": label.get("name", name),
"color": label.get("color", color),
},
},
status_code=201,
)
@app.delete("/api/repos/{owner}/{repo}/labels/{label_id}")
async def api_delete_label(
request: Request, owner: str, repo: str, label_id: int
):
"""Delete a label definition (CSRF via the X-CSRF-Token header)."""
if not _csrf_valid(request, ""):
return _csrf_error()
await gitea.delete_label(owner, repo, label_id)
return {"ok": True, "deleted": label_id}
@app.post("/api/issues/{owner}/{repo}/{number}/labels")
async def api_toggle_issue_label(
request: Request,
owner: str,
repo: str,
number: int,
label_id: int = Form(...),
action: str = Form(...),
csrf_token: str = Form(""),
):
"""Add or remove one label on one issue; returns the new label set."""
if not _csrf_valid(request, csrf_token):
return _csrf_error()
if action not in ("add", "remove"):
return JSONResponse(
{"ok": False, "error": "action must be 'add' or 'remove'"},
status_code=400,
)
if action == "add":
await gitea.add_issue_label(owner, repo, number, label_id)
else:
await gitea.remove_issue_label(owner, repo, number, label_id)
issue = await gitea.get_issue(owner, repo, number)
return {
"ok": True,
"labels": [
{"id": l.get("id"), "name": l.get("name", ""), "color": l.get("color", "")}
for l in issue.get("labels") or []
],
}
# ── MCP server (read-only tools over JSON-RPC) ─────────────
MCP_TOOLS = [
{
"name": "list_projects",
"description": "List the projects (Gitea repos) Kanbante tracks, grouped by activity.",
"inputSchema": {"type": "object", "properties": {}},
},
{
"name": "get_board",
"description": "Get the kanban board of one repo: columns with their cards.",
"inputSchema": {
"type": "object",
"properties": {
"owner": {"type": "string"},
"repo": {"type": "string"},
},
"required": ["owner", "repo"],
},
},
{
"name": "get_issue",
"description": "Get one issue with its body and comments.",
"inputSchema": {
"type": "object",
"properties": {
"owner": {"type": "string"},
"repo": {"type": "string"},
"number": {"type": "integer"},
},
"required": ["owner", "repo", "number"],
},
},
{
"name": "get_focus",
"description": "Cards in progress or in review across all active projects.",
"inputSchema": {"type": "object", "properties": {}},
},
{
"name": "search_issues",
"description": "Search issues by text across all repos.",
"inputSchema": {
"type": "object",
"properties": {
"query": {"type": "string"},
"state": {"type": "string", "enum": ["open", "closed", "all"]},
},
"required": ["query"],
},
},
{
"name": "get_activity",
"description": "Latest entries of the board event journal.",
"inputSchema": {
"type": "object",
"properties": {"limit": {"type": "integer"}},
},
},
{
"name": "get_metrics",
"description": "Throughput and cycle-time metrics from the journal.",
"inputSchema": {
"type": "object",
"properties": {"repo": {"type": "string"}},
},
},
]
async def _mcp_call_tool(name: str, args: dict) -> str:
"""Execute one read-only MCP tool and return text content."""
if name == "list_projects":
repos = await gitea.search_repos()
active, inactive, archived = classify_repos(repos)
lines = [f"Active projects ({len(active)}):"]
lines += [
f"- {r['full_name']} ({r.get('open_issues_count') or 0} open issues)"
for r in active
]
lines.append(
f"Inactive: {len(inactive)} - Archived: {len(archived)}"
)
return "\n".join(lines)
if name == "get_board":
owner, repo = args["owner"], args["repo"]
items = await gitea.get_repo_issues(owner, repo, state="all")
issues = [i for i in items if _is_issue(i)]
open_issues = [i for i in issues if i.get("state") == "open"]
columns = _merge_issues_with_board(open_issues, get_board_state(f"{owner}/{repo}"))
return json.dumps(
{
column: [
{
"number": i["number"],
"title": i["title"],
"labels": [l.get("name") for l in i.get("labels") or []],
}
for i in cards
]
for column, cards in columns.items()
},
ensure_ascii=False,
)
if name == "get_issue":
detail = await api_issue_detail(args["owner"], args["repo"], int(args["number"]))
return json.dumps(detail, ensure_ascii=False, default=str)
if name == "get_focus":
groups = await _collect_focus_groups()
return json.dumps(
[
{
"repo": g["full_name"],
"cards": [
{
"number": c["issue"]["number"],
"title": c["issue"]["title"],
"column": c["column"],
}
for c in g["cards"]
],
}
for g in groups
],
ensure_ascii=False,
)
if name == "search_issues":
results = await gitea.search_issues(
q=args["query"], state=args.get("state", "open")
)
return json.dumps(
[
{
"repo": (i.get("repository") or {}).get("full_name"),
"number": i["number"],
"title": i["title"],
"state": i["state"],
}
for i in results
],
ensure_ascii=False,
)
if name == "get_activity":
return json.dumps(
get_events(limit=int(args.get("limit", 50))), ensure_ascii=False
)
if name == "get_metrics":
return json.dumps(get_metrics(args.get("repo") or None), ensure_ascii=False)
raise ValueError(f"Unknown tool: {name}")
@app.get("/mcp")
async def mcp_info():
"""Discoverability for the MCP endpoint (the protocol is POST)."""
return {
"name": "kanbante",
"version": APP_VERSION,
"transport": "streamable-http (JSON-RPC 2.0 over POST)",
"read_only": True,
"tools": [t["name"] for t in MCP_TOOLS],
}
@app.post("/mcp")
async def mcp_endpoint(request: Request):
"""Minimal MCP server: initialize / tools/list / tools/call / ping.
Read-only by design (Bruno's rule: cross-platform reads first,
writes go through approval flows elsewhere).
"""
try:
payload = await request.json()
except ValueError:
return JSONResponse(
{"jsonrpc": "2.0", "id": None,
"error": {"code": -32700, "message": "Parse error"}},
status_code=400,
)
if isinstance(payload, list): # batch: not supported by this server
return JSONResponse(
{"jsonrpc": "2.0", "id": None,
"error": {"code": -32600, "message": "Batch requests not supported"}},
status_code=400,
)
method = payload.get("method", "")
rpc_id = payload.get("id")
if rpc_id is None: # notification: acknowledge silently
return Response(status_code=202)
params = payload.get("params") or {}
def _result(result):
return {"jsonrpc": "2.0", "id": rpc_id, "result": result}
def _error(code, message):
return {"jsonrpc": "2.0", "id": rpc_id,
"error": {"code": code, "message": message}}
if method == "initialize":
return _result(
{
"protocolVersion": params.get("protocolVersion", "2024-11-05"),
"capabilities": {"tools": {}},
"serverInfo": {"name": "kanbante", "version": APP_VERSION},
}
)
if method == "ping":
return _result({})
if method == "tools/list":
return _result({"tools": MCP_TOOLS})
if method == "tools/call":
tool_name = params.get("name", "")
arguments = params.get("arguments") or {}
try:
text = await _mcp_call_tool(tool_name, arguments)
except KeyError as exc:
return _result(
{"content": [{"type": "text", "text": f"Missing argument: {exc}"}],
"isError": True}
)
except Exception as exc:
return _result(
{"content": [{"type": "text", "text": f"Tool error: {exc}"}],
"isError": True}
)
return _result({"content": [{"type": "text", "text": text}], "isError": False})
return _error(-32601, f"Method not found: {method}")
@app.post("/api/notes")
async def api_notes(
request: Request,
+22
View File
@@ -79,6 +79,11 @@ def init_db() -> None:
CREATE INDEX IF NOT EXISTS idx_board_repo
ON board_state(repo_full_name, column_name);
CREATE TABLE IF NOT EXISTS kv_store (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_events_repo
ON board_events(repo_full_name, issue_number, created_at);
"""
@@ -234,6 +239,23 @@ def reconcile_board_state(
return changed
def get_kv(key: str) -> str | None:
with get_db() as conn:
row = conn.execute(
"SELECT value FROM kv_store WHERE key = ?", (key,)
).fetchone()
return row["value"] if row else None
def set_kv(key: str, value: str) -> None:
with get_db() as conn:
conn.execute(
"INSERT INTO kv_store (key, value) VALUES (?, ?) "
"ON CONFLICT(key) DO UPDATE SET value = excluded.value",
(key, value),
)
# ── Event journal ──────────────────────────────────────────
EVENT_TYPES = ("created", "moved", "closed", "reopened", "commented")