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")