76 lines
2.6 KiB
Python
76 lines
2.6 KiB
Python
from dataclasses import dataclass
|
|
from typing import List, Optional
|
|
from pathlib import Path
|
|
import json
|
|
|
|
@dataclass
|
|
class Bead:
|
|
id: str
|
|
title: str
|
|
description: str
|
|
status: str = "active"
|
|
|
|
class BeadsClient:
|
|
def __init__(self, working_dir: Path):
|
|
"""
|
|
[C: src/mcp_client.py:_DDGParser.__init__, src/mcp_client.py:_TextExtractor.__init__]
|
|
"""
|
|
self.working_dir = Path(working_dir)
|
|
self.repo_dir = self.working_dir / ".beads_mock"
|
|
self.beads_file = self.repo_dir / "beads.json"
|
|
|
|
def init_repo(self) -> None:
|
|
"""
|
|
Initialize the mock repository.
|
|
[C: tests/test_aggregate_beads.py:test_build_beads_compaction, tests/test_beads_client.py:test_beads_init_and_query, tests/test_gui_dag_beads.py:test_load_active_tickets_from_beads, tests/test_mcp_client_beads.py:test_bd_mcp_tools]
|
|
"""
|
|
self.repo_dir.mkdir(parents=True, exist_ok=True)
|
|
if not self.beads_file.exists():
|
|
self.beads_file.write_text("[]", encoding="utf-8")
|
|
|
|
def is_initialized(self) -> bool:
|
|
"""
|
|
Check if the repository is initialized.
|
|
[C: src/mcp_client.py:dispatch, tests/test_beads_client.py:test_beads_init_and_query]
|
|
"""
|
|
return self.beads_file.exists()
|
|
|
|
def create_bead(self, title: str, description: str) -> str:
|
|
"""
|
|
Create a new bead and return its ID.
|
|
[C: src/mcp_client.py:dispatch, tests/test_aggregate_beads.py:test_build_beads_compaction, tests/test_beads_client.py:test_beads_init_and_query, tests/test_gui_dag_beads.py:test_load_active_tickets_from_beads]
|
|
"""
|
|
beads = self._read_beads()
|
|
bead_id = f"bead-{len(beads) + 1}"
|
|
bead = {"id": bead_id, "title": title, "description": description, "status": "active"}
|
|
beads.append(bead)
|
|
self._write_beads(beads)
|
|
return bead_id
|
|
|
|
def update_bead(self, bead_id: str, status: str) -> bool:
|
|
"""
|
|
Update the status of an existing bead.
|
|
[C: src/mcp_client.py:dispatch, tests/test_aggregate_beads.py:test_build_beads_compaction, tests/test_beads_client.py:test_beads_init_and_query]
|
|
"""
|
|
beads = self._read_beads()
|
|
for bead in beads:
|
|
if bead["id"] == bead_id:
|
|
bead["status"] = status
|
|
self._write_beads(beads)
|
|
return True
|
|
return False
|
|
|
|
def list_beads(self) -> List[Bead]:
|
|
"""
|
|
List all beads.
|
|
[C: src/gui_2.py:App._render_beads_tab, src/mcp_client.py:dispatch, tests/test_beads_client.py:test_beads_init_and_query]
|
|
"""
|
|
return [Bead(**b) for b in self._read_beads()]
|
|
|
|
def _read_beads(self) -> List[dict]:
|
|
if not self.beads_file.exists():
|
|
return []
|
|
return json.loads(self.beads_file.read_text(encoding="utf-8"))
|
|
|
|
def _write_beads(self, beads: List[dict]) -> None:
|
|
self.beads_file.write_text(json.dumps(beads, indent=1), encoding="utf-8") |