221 lines
7.4 KiB
Python
221 lines
7.4 KiB
Python
from fastapi import FastAPI, Header, HTTPException, Request
|
|
from fastapi.staticfiles import StaticFiles
|
|
import httpx
|
|
from typing import Annotated
|
|
import asyncio
|
|
import time
|
|
|
|
app = FastAPI()
|
|
|
|
DECK_APIS = {
|
|
"prod": "https://deck-api.ngcloud.ru/api/v1",
|
|
"dev": "https://deck-api-dev.ngcloud.ru/api/v1",
|
|
"test": "https://deck-api-test.ngcloud.ru/api/v1",
|
|
}
|
|
DEFAULT_DECK_ENV = "test"
|
|
CACHE_TTL_SECONDS = 180
|
|
GRAPH_CACHE: dict[str, tuple[float, dict]] = {}
|
|
|
|
|
|
def auth_headers(token: str):
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def resolve_deck_api(env_raw: str | None) -> tuple[str, str]:
|
|
env = (env_raw or DEFAULT_DECK_ENV).strip().lower()
|
|
if env not in DECK_APIS:
|
|
env = DEFAULT_DECK_ENV
|
|
return env, DECK_APIS[env]
|
|
|
|
|
|
async def fetch_all_instances(token: str, deck_api: str):
|
|
params: dict = {"page": 1, "size": 200}
|
|
all_results = []
|
|
async with httpx.AsyncClient(timeout=20) as client:
|
|
while True:
|
|
r = await client.get(
|
|
f"{deck_api}/index.cfm/instances",
|
|
headers=auth_headers(token),
|
|
params=params,
|
|
)
|
|
if r.status_code == 401:
|
|
raise HTTPException(status_code=401, detail="Invalid token")
|
|
data = r.json()
|
|
results = data.get("results", [])
|
|
all_results.extend(results)
|
|
if len(results) < 200:
|
|
break
|
|
params["page"] += 1
|
|
return all_results
|
|
|
|
|
|
@app.get("/api/instances")
|
|
async def list_instances(
|
|
request: Request,
|
|
x_deck_token: Annotated[str | None, Header()] = None,
|
|
x_deck_env: Annotated[str | None, Header()] = None,
|
|
):
|
|
if not x_deck_token:
|
|
raise HTTPException(status_code=401, detail="Token required")
|
|
_, deck_api = resolve_deck_api(x_deck_env)
|
|
# parse status from query string manually (FastAPI list param can fail behind ingress rewrite)
|
|
status_list = request.query_params.getlist("status")
|
|
all_results = await fetch_all_instances(x_deck_token, deck_api)
|
|
if status_list:
|
|
all_results = [i for i in all_results if i.get("explainedStatus") in status_list]
|
|
return {"instances": all_results, "total": len(all_results)}
|
|
|
|
|
|
@app.get("/api/graph")
|
|
async def get_graph(
|
|
request: Request,
|
|
x_deck_token: Annotated[str | None, Header()] = None,
|
|
x_deck_env: Annotated[str | None, Header()] = None,
|
|
):
|
|
if not x_deck_token:
|
|
raise HTTPException(status_code=401, detail="Token required")
|
|
env, deck_api = resolve_deck_api(x_deck_env)
|
|
|
|
status_list = request.query_params.getlist("status")
|
|
root_uid = request.query_params.get("root_uid")
|
|
depth_raw = request.query_params.get("depth", "2")
|
|
try:
|
|
depth = max(1, min(5, int(depth_raw)))
|
|
except ValueError:
|
|
depth = 2
|
|
|
|
status_key = ",".join(sorted(status_list)) if status_list else "__all__"
|
|
cache_key = f"{x_deck_token[:24]}::{env}::{status_key}"
|
|
now = time.time()
|
|
|
|
cached = GRAPH_CACHE.get(cache_key)
|
|
if cached and (now - cached[0] < CACHE_TTL_SECONDS):
|
|
base_graph = cached[1]
|
|
else:
|
|
all_instances = await fetch_all_instances(x_deck_token, deck_api)
|
|
if status_list:
|
|
all_instances = [i for i in all_instances if i.get("explainedStatus") in status_list]
|
|
|
|
uid_to_inst = {
|
|
i.get("instanceUid"): i
|
|
for i in all_instances
|
|
if i.get("instanceUid")
|
|
}
|
|
allowed_uids = set(uid_to_inst.keys())
|
|
|
|
sem = asyncio.Semaphore(16)
|
|
|
|
async def fetch_detail(uid: str):
|
|
async with sem:
|
|
async with httpx.AsyncClient(timeout=20) as client:
|
|
r = await client.get(
|
|
f"{deck_api}/index.cfm/instances/{uid}",
|
|
headers=auth_headers(x_deck_token),
|
|
)
|
|
if r.status_code == 401:
|
|
raise HTTPException(status_code=401, detail="Invalid token")
|
|
if r.status_code >= 400:
|
|
return uid, None
|
|
return uid, r.json().get("instance")
|
|
|
|
details = await asyncio.gather(*[fetch_detail(uid) for uid in allowed_uids])
|
|
detail_map = {uid: inst for uid, inst in details if inst}
|
|
|
|
edge_set = set()
|
|
|
|
def dep_uid(dep_obj: dict):
|
|
return dep_obj.get("uid") or dep_obj.get("instanceUid")
|
|
|
|
for uid, inst in detail_map.items():
|
|
for dep in inst.get("dependencies", []) or []:
|
|
src = dep_uid(dep)
|
|
if src and src in allowed_uids:
|
|
edge_set.add((src, uid))
|
|
for dep_on in inst.get("dependentInstances", []) or []:
|
|
dst = dep_uid(dep_on)
|
|
if dst and dst in allowed_uids:
|
|
edge_set.add((uid, dst))
|
|
|
|
nodes = []
|
|
for uid in allowed_uids:
|
|
i = uid_to_inst.get(uid, {})
|
|
nodes.append(
|
|
{
|
|
"id": uid,
|
|
"label": i.get("displayName") or uid,
|
|
"status": i.get("explainedStatus") or "unknown",
|
|
"svc": i.get("svc") or "",
|
|
"realm": i.get("resourceRealm") or "",
|
|
}
|
|
)
|
|
|
|
edges = [{"source": s, "target": t} for s, t in sorted(edge_set)]
|
|
base_graph = {"nodes": nodes, "edges": edges}
|
|
GRAPH_CACHE[cache_key] = (now, base_graph)
|
|
|
|
# Return full graph fast
|
|
if not root_uid:
|
|
return {
|
|
"nodes": base_graph["nodes"],
|
|
"edges": base_graph["edges"],
|
|
"total_nodes": len(base_graph["nodes"]),
|
|
"total_edges": len(base_graph["edges"]),
|
|
"cached": bool(cached),
|
|
}
|
|
|
|
node_ids = {n["id"] for n in base_graph["nodes"]}
|
|
if root_uid not in node_ids:
|
|
return {"nodes": [], "edges": [], "total_nodes": 0, "total_edges": 0, "cached": bool(cached)}
|
|
|
|
adj: dict[str, set[str]] = {uid: set() for uid in node_ids}
|
|
for e in base_graph["edges"]:
|
|
s, t = e["source"], e["target"]
|
|
if s in adj and t in adj:
|
|
adj[s].add(t)
|
|
adj[t].add(s)
|
|
|
|
frontier = {root_uid}
|
|
seen = {root_uid}
|
|
for _ in range(depth):
|
|
nxt = set()
|
|
for node in frontier:
|
|
nxt |= adj.get(node, set())
|
|
nxt -= seen
|
|
if not nxt:
|
|
break
|
|
seen |= nxt
|
|
frontier = nxt
|
|
|
|
sub_nodes = [n for n in base_graph["nodes"] if n["id"] in seen]
|
|
sub_edges = [e for e in base_graph["edges"] if e["source"] in seen and e["target"] in seen]
|
|
return {
|
|
"nodes": sub_nodes,
|
|
"edges": sub_edges,
|
|
"total_nodes": len(sub_nodes),
|
|
"total_edges": len(sub_edges),
|
|
"cached": bool(cached),
|
|
}
|
|
|
|
|
|
@app.get("/api/instances/{uid}")
|
|
async def get_instance(
|
|
request: Request,
|
|
uid: str,
|
|
x_deck_token: Annotated[str | None, Header()] = None,
|
|
x_deck_env: Annotated[str | None, Header()] = None,
|
|
):
|
|
if not x_deck_token:
|
|
raise HTTPException(status_code=401, detail="Token required")
|
|
_, deck_api = resolve_deck_api(x_deck_env)
|
|
async with httpx.AsyncClient(timeout=20) as client:
|
|
r = await client.get(
|
|
f"{deck_api}/index.cfm/instances/{uid}",
|
|
headers=auth_headers(x_deck_token),
|
|
)
|
|
if r.status_code == 401:
|
|
raise HTTPException(status_code=401, detail="Invalid token")
|
|
return r.json()
|
|
|
|
|
|
app.mount("/", StaticFiles(directory="/app/static", html=True), name="static")
|