108 lines
3.9 KiB
Python
108 lines
3.9 KiB
Python
import asyncio
|
|
import math
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from src.gitea_proxy import activity_events, current_user, repos, issues, pull_requests
|
|
from src.models import User, Repo, Issue, PullRequest
|
|
from src.suggestion_engine import compute
|
|
from src.views import router as frontend_router
|
|
|
|
app = FastAPI(title="Stackchain Dashboard")
|
|
CONTEXT_TIMEOUT_SECONDS = 5.0
|
|
FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=False,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
app.mount("/static", StaticFiles(directory=FRONTEND_DIR), name="static")
|
|
app.include_router(frontend_router)
|
|
|
|
|
|
@app.get("/healthz")
|
|
def health() -> dict[str, str]:
|
|
"""Return process liveness without depending on Gitea."""
|
|
return {"status": "ok", "service": "stackchain-dashboard"}
|
|
|
|
|
|
@app.get("/readyz")
|
|
async def readiness():
|
|
"""Return readiness after verifying the configured Gitea connection."""
|
|
try:
|
|
user = await current_user()
|
|
except Exception as exc:
|
|
return JSONResponse(
|
|
{
|
|
"status": "not_ready",
|
|
"service": "stackchain-dashboard",
|
|
"error": str(exc),
|
|
},
|
|
status_code=503,
|
|
)
|
|
return {
|
|
"status": "ready",
|
|
"service": "stackchain-dashboard",
|
|
"gitea_user": user["login"],
|
|
}
|
|
|
|
|
|
@app.get("/api/v1/context")
|
|
async def context() -> JSONResponse:
|
|
try:
|
|
user_data, repo_data, issues_data, prs_data = await asyncio.wait_for(
|
|
asyncio.gather(current_user(), repos(), issues(), pull_requests()),
|
|
timeout=CONTEXT_TIMEOUT_SECONDS,
|
|
)
|
|
except Exception as e:
|
|
error_message = (
|
|
f"Gitea context request timed out after {CONTEXT_TIMEOUT_SECONDS:g}s"
|
|
if isinstance(e, TimeoutError)
|
|
else str(e)
|
|
)
|
|
return JSONResponse({
|
|
"user": {"id": None, "login": "timmy", "full_name": "Timmy Jr", "email": ""},
|
|
"repos": [],
|
|
"issues": [],
|
|
"pull_requests": [],
|
|
"view": "dashboard",
|
|
"deltas": [
|
|
{"panel": "auth", "action": "connect", "target": "stackchain-dashboard backend", "priority": "high"},
|
|
{"panel": "gitea", "action": "set GITEA_URL + GITEA_TOKEN", "target": "/root/stackchain-dashboard/.env or systemd env", "priority": "high"},
|
|
],
|
|
"error": error_message,
|
|
}, headers={
|
|
"Retry-After": str(max(1, math.ceil(CONTEXT_TIMEOUT_SECONDS)))
|
|
} if isinstance(e, TimeoutError) else None)
|
|
user_model = User(id=user_data["id"], login=user_data["login"], full_name=user_data.get("full_name", ""), email=user_data.get("email", ""))
|
|
repo_models = [
|
|
Repo(id=r["id"], name=r["name"], full_name=r["full_name"], description=r.get("description") or "", url=r["html_url"], updated_at=r.get("updated_at", ""))
|
|
for r in repo_data[:50]
|
|
if isinstance(r, dict)
|
|
]
|
|
issue_models = [
|
|
Issue(id=i["id"], number=i["number"], title=i["title"], state=i["state"], labels=[l.get("name", "") for l in (i.get("labels") or [])], assignees=[a.get("login", "") for a in (i.get("assignees") or [])], url=i["html_url"])
|
|
for i in issues_data[:50]
|
|
if isinstance(i, dict)
|
|
]
|
|
pr_models = [
|
|
PullRequest(id=p["id"], number=p["number"], title=p["title"], state=p["state"], user=(p.get("user") or {}).get("login", ""), url=p["html_url"])
|
|
for p in prs_data[:50]
|
|
if isinstance(p, dict)
|
|
]
|
|
ctx = compute(user_model, repo_models, issue_models, pr_models)
|
|
return JSONResponse(ctx.model_dump())
|
|
|
|
|
|
@app.get("/api/v1/events")
|
|
async def event_stream() -> list[dict]:
|
|
return await activity_events()
|