(self, routes: Dict[str, ApiHandler], body: Dict[str, Any] | None = None)
| 513 | self.wfile.write(data) |
| 514 | |
| 515 | def _handle_api(self, routes: Dict[str, ApiHandler], body: Dict[str, Any] | None = None) -> None: |
| 516 | path, query_params = _query(self.path) |
| 517 | route = routes.get(path) |
| 518 | if route is None: |
| 519 | self._send_json({"ok": False, "error": f"Unknown API route: {path}"}, status=HTTPStatus.NOT_FOUND) |
| 520 | return |
| 521 | try: |
| 522 | payload = route(query_params, body or {}) |
| 523 | self._send_json({"ok": True, **payload}) |
| 524 | except Exception as exc: |
| 525 | self._send_json({"ok": False, "error": str(exc)}, status=HTTPStatus.BAD_REQUEST) |
| 526 | |
| 527 | def _send_sse_event(self, event: str, payload: Dict[str, Any]) -> None: |
| 528 | self.wfile.write(_sse_bytes(event, payload)) |
no test coverage detected