import json import os import subprocess import time from pathlib import Path from fastapi import FastAPI, HTTPException from pydantic import BaseModel import uvicorn app = FastAPI(title="Stateless ClickHouse Gateway") USER_FILES_DIR = "/app/ch/user_files" SQL_DIR = "/app/sql_repo" def sync_git_repo(repo_url, target_dir): """Clones or forcefully updates a git repository.""" os.makedirs(target_dir, exist_ok=True) git_dir = os.path.join(target_dir, ".git") if not os.path.exists(git_dir): print(f"Cloning {repo_url} into {target_dir}...") subprocess.run(["git", "clone", repo_url, target_dir], check=True) else: print(f"Force updating {repo_url} in {target_dir}...") # Fetch all remote changes and forcefully reset to avoid merge conflicts subprocess.run(["git", "-C", target_dir, "fetch", "--all"], check=True) subprocess.run(["git", "-C", target_dir, "reset", "--hard", "FETCH_HEAD"], check=True) def setup_database(): """Pulls data/SQL and executes nested SQL scripts.""" with open("/app/config.json") as f: config = json.load(f) # 1. Sync Data Repositories for i, repo_url in enumerate(config.get("data_repos", [])): repo_dir = os.path.join(USER_FILES_DIR, f"repo_{i}") sync_git_repo(repo_url, repo_dir) # 2. Sync SQL Repository sync_git_repo(config.get("sql_repo"), SQL_DIR) # 3. Find and Execute SQL scripts recursively (handles nested folders) print("--- SCANNING FOR SQL SCRIPTS ---") # rglob finds .sql files in all subdirectories. sorted() ensures deterministic execution. sql_paths = sorted(Path(SQL_DIR).rglob("*.sql")) if not sql_paths: print("No .sql files found in the repository!") for path in sql_paths: print(f"Executing {path.relative_to(SQL_DIR)}...") subprocess.run(f"clickhouse-client --queries-file {str(path)}", shell=True, check=True) @app.on_event("startup") def startup_event(): print("--- STARTING CLICKHOUSE ---") subprocess.Popen(["clickhouse-server", "--config-file=/etc/clickhouse-server/config.xml"]) # Wait for ClickHouse to boot ready = False for _ in range(45): if subprocess.run(["clickhouse-client", "--query", "SELECT 1"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode == 0: ready = True break time.sleep(1) if not ready: raise Exception("CRITICAL: ClickHouse failed to start. Container crashing.") print("--- CLICKHOUSE IS ALIVE. ATTEMPTING DATA SYNC ---") try: setup_database() print("--- DATA SYNC COMPLETE ---") except Exception as e: # Graceful degradation: If git or SQL fails, log it but keep DB running print(f"WARNING: Data sync or SQL execution failed: {e}") print("ClickHouse is still running and available for manual queries.") print("--- SYSTEM READY ---") @app.post("/refresh") def refresh_data(): """Endpoint to pull the latest Git changes and re-run SQL.""" try: setup_database() return {"status": "success", "message": "Git repositories updated and SQL re-executed."} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) class QueryRequest(BaseModel): sql: str @app.post("/query") def run_query(req: QueryRequest): """Endpoint to execute raw SQL against ClickHouse.""" process = subprocess.run( ["clickhouse-client", "--query", req.sql, "--format", "JSON"], capture_output=True, text=True ) if process.returncode != 0: raise HTTPException(status_code=400, detail=process.stderr) return json.loads(process.stdout) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=7860)