Spaces:
Sleeping
Sleeping
File size: 2,426 Bytes
3fd3eea | 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 | from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse, HTMLResponse
import subprocess
import httpx
app = FastAPI(title="Market-Data Observatory")
# Async HTTP client for proxying to ClickHouse's internal HTTP interface
ch_client = httpx.AsyncClient(base_url="http://127.0.0.1:8123", timeout=60.0)
# ── Refresh endpoint (git pull) ──
@app.post("/refresh")
def refresh():
try:
result = subprocess.check_output(
"cd /app/data/ohlc_data && git pull",
shell=True, stderr=subprocess.STDOUT
)
return {"status": "updated", "details": result.decode().strip()}
except subprocess.CalledProcessError as e:
msg = e.output.decode() if e.output else str(e)
raise HTTPException(status_code=500, detail={"error": "Refresh failed", "message": msg})
# ── Health check ──
@app.get("/health")
async def health():
try:
r = await ch_client.get("/ping")
return {"clickhouse": r.text.strip(), "status": "ok"}
except Exception as e:
raise HTTPException(status_code=503, detail={"status": "unhealthy", "error": str(e)})
# ── Catch-all reverse proxy to ClickHouse ──
# This gives you /play, /dashboard, native HTTP API, everything.
@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "OPTIONS", "HEAD", "PATCH"])
async def proxy_to_clickhouse(request: Request, path: str):
url = httpx.URL(path=f"/{path}", query=request.url.query.encode("utf-8"))
# Forward headers, strip hop-by-hop
headers = dict(request.headers)
for h in ("host", "content-length", "transfer-encoding"):
headers.pop(h, None)
try:
body = await request.body()
req = ch_client.build_request(
method=request.method,
url=url,
headers=headers,
content=body,
)
r = await ch_client.send(req, stream=True)
# Pass through ClickHouse response headers
resp_headers = dict(r.headers)
for h in ("content-length", "content-encoding", "transfer-encoding"):
resp_headers.pop(h, None)
return StreamingResponse(
r.aiter_raw(),
status_code=r.status_code,
headers=resp_headers,
)
except httpx.RequestError as exc:
raise HTTPException(status_code=502, detail=f"ClickHouse proxy error: {exc}")
|