import asyncio import logging import transports import uvicorn from pydantic import BaseModel, Field from spaday import CallEndpoint, element, event_value from spaday.backends.starlette import serve from starlette.responses import JSONResponse from starlette.routing import Route, WebSocketRoute from spaday_trees import Tree, package logger = logging.getLogger("uvicorn.error") initial_paths = [ ".github/workflows/build.yaml", ".github/workflows/docs.yaml", "docs/src/explanation.md", "docs/src/how-to.md", "docs/src/reference.md", "docs/src/tutorial.md", "js/src/ts/index.ts", "js/tests/index.spec.js", "spaday_trees/__init__.py", "spaday_trees/components.cem.json", "spaday_trees/components.py", "spaday_trees/tests/test_all.py", "AGENTS.md", "LICENSE", "README.md", "spaday_trees/example.py", "pyproject.toml", ] class ProjectFeed(BaseModel): paths: list[str] = Field(default_factory=lambda: list(initial_paths)) selected_paths: list[str] = Field(default_factory=lambda: ["README.md"]) search: str = "" file_count: str = str(len(initial_paths)) changed_count: str = "5" sync_status: str = "Connected" feed = ProjectFeed() session = transports.Session() session.host(feed) server = transports.Server(session) git_status = [ {"path": "README.md", "status": "modified"}, {"path": "spaday_trees/example.py", "status": "added"}, {"path": "docs/src/tutorial.md", "status": "modified"}, {"path": "js/tests/index.spec.js", "status": "modified"}, {"path": "spaday_trees/components.py", "status": "untracked"}, ] async def rotate_server_file() -> None: tick = 0 generated: str | None = None while True: await asyncio.sleep(5) tick += 1 paths = [path for path in feed.paths if path != generated] generated = f"server/review-{tick:02d}.md" feed.paths = [*paths, generated] feed.file_count = str(len(feed.paths)) feed.sync_status = f"Server added {generated}" async def tree_event(request): event_type = request.path_params["event_type"] payload = await request.json() if event_type == "selection": paths = [str(path) for path in payload.get("paths", [])] feed.selected_paths = paths message = f"Selected {paths[0]}" if paths else "Selection cleared" elif event_type == "search": value = str(payload.get("value") or "") feed.search = value message = f"Searching for “{value}”" if value else "Search cleared" else: return JSONResponse({"message": "Unknown tree event"}, status_code=404) feed.sync_status = message logger.info("Tree %s received from browser: %s", event_type, payload) return JSONResponse({"message": message}) tree = ( Tree( paths=initial_paths, selected_paths=["README.md"], git_status=git_status, id="project-tree", ) .bind("paths", "paths") .bind("selected_paths", "selected_paths") .bind("search", "search") .on("selection-change", CallEndpoint("POST", "/api/tree/selection", event_value(), result="tree_result")) .on("search-change", CallEndpoint("POST", "/api/tree/search", event_value(), result="tree_result")) ) page = element( "main", element( "header", element( "div", element("p", class_="eyebrow").text("REACTIVE PROJECT EXPLORER"), element("h1").text("Pierre Trees workspace"), element("p", class_="lede").text("A fast file tree with search, selection, Git status, and bidirectional server sync."), ), element("div", element("span", class_="pulse"), element("strong").bind("textContent", "sync_status"), class_="connection"), class_="page-header", ), element( "section", element("article", element("span").text("Project files"), element("strong").bind("textContent", "file_count")), element("article", element("span").text("Changed files"), element("strong").bind("textContent", "changed_count")), element("article", element("span").text("Active branch"), element("strong").text("tkp/examples")), class_="metrics", ), element( "section", element( "aside", element( "div", element("span", class_="window-dot red"), element("span", class_="window-dot yellow"), element("span", class_="window-dot green"), class_="window-controls", ), element("div", element("strong").text("EXPLORER"), element("span").text("SPADAY-TREES"), class_="panel-title"), tree, class_="tree-panel", ), element( "article", element("span", class_="file-label").text("README.md"), element("h2").text("Build interfaces from typed components"), element("p").text("Select any item in the tree. Selection and search events return to Python and are logged by the server."), element( "div", element("button") .text("Select README") .on( "click", CallEndpoint("POST", "/api/tree/selection", {"paths": ["README.md"]}, result="tree_result"), ), element("button") .text("Select component") .on( "click", CallEndpoint( "POST", "/api/tree/selection", {"paths": ["spaday_trees/components.py"]}, result="tree_result", ), ), element("button") .text("Clear selection") .on("click", CallEndpoint("POST", "/api/tree/selection", {"paths": []}, result="tree_result")), class_="actions", ), element( "div", element("span", class_="status-dot"), element("span").bind("textContent", "tree_result.body.message"), class_="event-status" ), element( "div", element("span").text("M"), element("code").text("README.md"), element("span").text("Modified"), class_="change-row", ), element( "div", element("span").text("A"), element("code").text("spaday_trees/example.py"), element("span").text("Added"), class_="change-row added", ), class_="preview-panel", ), class_="workspace", ), element("p", class_="footnote").text( "The server replaces one generated file every five seconds; browser selection and search round-trip immediately." ), class_="page", ) styles = """ """ app = serve( page, packages=[package], wire="transports", routes=[ WebSocketRoute("/ws", transports.ws_endpoint(server)), Route("/api/tree/{event_type}", tree_event, methods=["POST"]), ], background=[transports.autosync(server), rotate_server_file()], head=styles, title="spaday-trees example", ) if __name__ == "__main__": uvicorn.run(app, host="127.0.0.1", port=8016)