Spaces:
Sleeping
Sleeping
File size: 3,821 Bytes
b8d96ef | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | 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) |