| 2122 | page = generate_dynamic_html_page() |
| 2123 | |
| 2124 | class Handler(BaseHTTPRequestHandler): |
| 2125 | def do_GET(self) -> None: # noqa: N802 |
| 2126 | try: |
| 2127 | self._dispatch() |
| 2128 | except Exception as exc: |
| 2129 | body = str(exc).encode("utf-8") |
| 2130 | self.send_response(500) |
| 2131 | self.send_header("Content-Type", "text/plain; charset=utf-8") |
| 2132 | self.send_header("Content-Length", str(len(body))) |
| 2133 | self.end_headers() |
| 2134 | self.wfile.write(body) |
| 2135 | |
| 2136 | def log_message(self, format: str, *args: Any) -> None: |
| 2137 | return |
| 2138 | |
| 2139 | def _dispatch(self) -> None: |
| 2140 | parsed = urllib.parse.urlparse(self.path) |
| 2141 | params = urllib.parse.parse_qs(parsed.query) |
| 2142 | |
| 2143 | if parsed.path in {"/", "/index.html"}: |
| 2144 | body = page.encode("utf-8") |
| 2145 | self.send_response(200) |
| 2146 | self.send_header("Content-Type", "text/html; charset=utf-8") |
| 2147 | self.send_header("Content-Length", str(len(body))) |
| 2148 | self.end_headers() |
| 2149 | self.wfile.write(body) |
| 2150 | return |
| 2151 | |
| 2152 | if parsed.path == "/api/meta": |
| 2153 | self._write_json(engine.meta()) |
| 2154 | return |
| 2155 | |
| 2156 | if parsed.path == "/api/threads": |
| 2157 | self._write_json(engine.list_threads()) |
| 2158 | return |
| 2159 | |
| 2160 | if parsed.path == "/api/functions": |
| 2161 | rows = engine.list_functions( |
| 2162 | thread_ids=_parse_csv_ints(params.get("threads", [""])[0]), |
| 2163 | query=params.get("q", [""])[0], |
| 2164 | limit=int(params.get("limit", ["200"])[0]), |
| 2165 | sort=params.get("sort", ["alpha"])[0], |
| 2166 | start_ms=int(params["start"][0]) if "start" in params else None, |
| 2167 | end_ms=int(params["end"][0]) if "end" in params else None, |
| 2168 | ) |
| 2169 | self._write_json(rows) |
| 2170 | return |
| 2171 | |
| 2172 | if parsed.path == "/api/series": |
| 2173 | thread_ids = _parse_csv_ints(params.get("threads", [""])[0]) |
| 2174 | function_ids = _parse_csv_ints(params.get("functions", [""])[0]) |
| 2175 | start_ms = int(params["start"][0]) if "start" in params else None |
| 2176 | end_ms = int(params["end"][0]) if "end" in params else None |
| 2177 | metric_default = params.get("value", ["malloc"])[0] |
| 2178 | selected_metrics = _parse_csv_strings(params.get("metrics", [metric_default])[0]) |
| 2179 | response = engine.query_multi_series( |
| 2180 | thread_ids=thread_ids, |
| 2181 | function_ids=function_ids, |
nothing calls this directly
no outgoing calls
no test coverage detected