#!/usr/bin/env python """Local HTTP API for the CodeRAG React demo.""" import argparse import json import mimetypes import os from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import urlparse from coderag import build_file_index from coderag.vector_db import CodeRagVectorDB ROOT = Path(__file__).resolve().parent MAX_REQUEST_BYTES = 2 * 1024 * 1024 STATIC_ROOT = ROOT / "coderag_ui" / "dist" DEFAULT_SAMPLE_CODE = """from collections import defaultdict def normalize_user(user): email = user["email"].strip().lower() return { "id": user["id"], "email": email, "active": bool(user.get("active", True)), } def group_active_users(users): groups = defaultdict(list) for user in users: normalized = normalize_user(user) if normalized["active"]: domain = normalized["email"].split("@")[-1] groups[domain].append(normalized) return dict(groups) def find_user_by_email(users, email): target = email.strip().lower() for user in users: normalized = normalize_user(user) if normalized["email"] == target: return normalized return None """ class CodeRagApiHandler(BaseHTTPRequestHandler): server_version = "CodeRAGApi/1.0" def log_message(self, format_string, *args): print(f"{self.address_string()} - {format_string % args}") @property def vector_db(self) -> CodeRagVectorDB: return self.server.vector_db def _send_json(self, status: int, payload: dict) -> None: body = json.dumps(payload).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.end_headers() self.wfile.write(body) def do_OPTIONS(self): self._send_json(204, {}) def do_GET(self): path = urlparse(self.path).path if path == "/api/health": self._send_json(200, {"status": "ok"}) return if path == "/api/coderag/stats": try: self._send_json(200, self.vector_db.stats()) except Exception as error: self._send_json( 500, {"error": f"Stats failed: {type(error).__name__}: {error}"}, ) return self._serve_static(path) def do_POST(self): path = urlparse(self.path).path if path not in { "/api/coderag/search", "/api/coderag/rebuild", "/api/coderag/index-code", }: self._send_json(404, {"error": "Not found"}) return try: payload = self._read_payload() except ValueError as error: self._send_json(400, {"error": str(error)}) return if path == "/api/coderag/rebuild": try: if self.server.repository_index.exists(): result = self.vector_db.build( repository_index_path=self.server.repository_index, synthetic_questions_path=self.server.synthetic_questions, ) else: sample_dir = ROOT / ".code_query_repo_cache" / "sample" sample_dir.mkdir(parents=True, exist_ok=True) sample_path = sample_dir / "sample_users.py" sample_path.write_text(DEFAULT_SAMPLE_CODE, encoding="utf-8") result = self.vector_db.build_from_file_index( build_file_index(sample_path) ) except Exception as error: self._send_json( 500, { "error": f"Rebuild failed: " f"{type(error).__name__}: {error}" }, ) return self._send_json(200, result) return if path == "/api/coderag/index-code": code = payload.get("code") filename = payload.get("filename", "pasted_code.py") if not isinstance(code, str) or not code.strip(): self._send_json(400, {"error": "'code' must be a non-empty string"}) return if not isinstance(filename, str) or not filename.strip(): filename = "pasted_code.py" if not filename.endswith(".py"): filename += ".py" safe_name = "".join( char if char.isalnum() or char in {"_", "-", "."} else "_" for char in filename ) pasted_dir = ROOT / ".code_query_repo_cache" / "pasted" pasted_dir.mkdir(parents=True, exist_ok=True) pasted_path = pasted_dir / safe_name try: pasted_path.write_text(code, encoding="utf-8") index = build_file_index( pasted_path, propagation_depth=int(payload.get("propagation_depth", 2)), max_callees=int(payload.get("max_callees", 4)), callee_context_tokens=int( payload.get("callee_context_tokens", 192) ), ) result = self.vector_db.build_from_file_index(index) except SyntaxError as error: self._send_json( 400, { "error": ( f"Python parse failed at line {error.lineno}: " f"{error.msg}" ) }, ) return except Exception as error: self._send_json( 500, { "error": f"Indexing failed: " f"{type(error).__name__}: {error}" }, ) return self._send_json(200, result) return query = payload.get("query") if not isinstance(query, str) or not query.strip(): self._send_json(400, {"error": "'query' must be a non-empty string"}) return try: result = self.vector_db.search( query, top_k=int(payload.get("top_k", 8)), per_source=int(payload.get("per_source", 5)), sources=payload.get("sources") or None, ) except Exception as error: self._send_json( 500, {"error": f"Search failed: {type(error).__name__}: {error}"}, ) return self._send_json(200, result) def _read_payload(self) -> dict: try: content_length = int(self.headers.get("Content-Length", "0")) except ValueError as error: raise ValueError("Invalid Content-Length") from error if content_length < 0 or content_length > MAX_REQUEST_BYTES: raise ValueError("Request body is too large") if content_length == 0: return {} try: return json.loads(self.rfile.read(content_length)) except (UnicodeDecodeError, json.JSONDecodeError) as error: raise ValueError("Request body must be valid JSON") from error def _serve_static(self, request_path: str) -> None: if not STATIC_ROOT.exists(): self._send_json(404, {"error": "UI build not found. Run npm run build."}) return relative = request_path.lstrip("/") or "index.html" candidate = (STATIC_ROOT / relative).resolve() try: candidate.relative_to(STATIC_ROOT.resolve()) except ValueError: self._send_json(403, {"error": "Forbidden"}) return if not candidate.exists() or candidate.is_dir(): candidate = STATIC_ROOT / "index.html" content_type = mimetypes.guess_type(candidate.name)[0] or "application/octet-stream" body = candidate.read_bytes() self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) class CodeRagApiServer(ThreadingHTTPServer): def __init__( self, server_address, handler_class, *, vector_db: CodeRagVectorDB, repository_index: Path, synthetic_questions: Path, ): super().__init__(server_address, handler_class) self.vector_db = vector_db self.repository_index = repository_index self.synthetic_questions = synthetic_questions def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Run the CodeRAG API server.") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=int(os.environ.get("PORT", "8010"))) parser.add_argument( "--database", type=Path, default=ROOT / ".code_query_repo_cache" / "coderag_vectors.sqlite", ) parser.add_argument( "--repository-index", type=Path, default=ROOT / "coderag_repository_index.json", ) parser.add_argument( "--synthetic-questions", type=Path, default=ROOT / "coderag_synthetic_queries.json", ) parser.add_argument( "--rebuild", action="store_true", help="Rebuild the SQLite vector database before serving.", ) return parser def main() -> int: args = build_parser().parse_args() vector_db = CodeRagVectorDB(args.database) vector_db.initialize() if args.rebuild or not args.database.exists(): stats = vector_db.build( repository_index_path=args.repository_index, synthetic_questions_path=args.synthetic_questions, ) print(f"Built vector database: {json.dumps(stats, indent=2)}", flush=True) server = CodeRagApiServer( (args.host, args.port), CodeRagApiHandler, vector_db=vector_db, repository_index=args.repository_index, synthetic_questions=args.synthetic_questions, ) url = f"http://{args.host}:{args.port}" print(f"CodeRAG API running at {url}", flush=True) print("Press Ctrl+C to stop.", flush=True) try: server.serve_forever() except KeyboardInterrupt: print("\nStopping CodeRAG API.") finally: server.server_close() return 0 if __name__ == "__main__": raise SystemExit(main())