From e835604a094ef362af92e4f5ae33e075d2f2051e Mon Sep 17 00:00:00 2001 From: haikdc Date: Sun, 5 Apr 2026 20:58:44 -0700 Subject: [PATCH] [Haik]: ckpt, initial version of NineRouter singelton --- .../subscriptions/NineRouter/NineRouter.py | 102 ++++++++++++++++++ .../NineRouter/NineRouterClient.py | 7 ++ backend/apps/subscriptions/subscriptions.py | 54 ++++------ 3 files changed, 130 insertions(+), 33 deletions(-) create mode 100644 backend/apps/subscriptions/NineRouter/NineRouter.py diff --git a/backend/apps/subscriptions/NineRouter/NineRouter.py b/backend/apps/subscriptions/NineRouter/NineRouter.py new file mode 100644 index 00000000..cd215945 --- /dev/null +++ b/backend/apps/subscriptions/NineRouter/NineRouter.py @@ -0,0 +1,102 @@ +"""Singleton that owns 9Router process lifecycle and client access. + +All interaction with 9Router from the subscriptions subapp goes through +NineRouter.get() so that mutable state (subprocess handle, background +ensure-task) lives in one place. +""" + +import asyncio +from typing import ClassVar, Optional + +from typeguard import typechecked + +from backend.apps.subscriptions.NineRouter.NineRouterProcess.NineRouterProcess import ( + is_running as _is_running, + ensure_running as _ensure_running, + stop as _stop, +) +from backend.apps.subscriptions.NineRouter.NineRouterClient import ( + get_providers as _get_providers, + get_models as _get_models, + start_oauth as _start_oauth, + poll_oauth as _poll_oauth, + exchange_oauth as _exchange_oauth, + disconnect_provider as _disconnect_provider, +) + + +class NineRouter: + _instance: ClassVar[Optional["NineRouter"]] = None + + def __init__(self) -> None: + self._ensure_task: Optional[asyncio.Task] = None + + @classmethod + def get(cls) -> "NineRouter": + if cls._instance is None: + cls._instance = cls() + return cls._instance + + # -- lifecycle ------------------------------------------------------------- + + @typechecked + def is_running(self) -> bool: + return _is_running() + + @typechecked + async def ensure_running(self) -> None: + await _ensure_running() + + @typechecked + def stop(self) -> None: + _stop() + + @typechecked + async def ensure_running_background(self) -> None: + """Kick off ensure_running as a background task if not already in flight.""" + if self._ensure_task is None or self._ensure_task.done(): + self._ensure_task = asyncio.create_task(_ensure_running()) + + @typechecked + def cancel_ensure_task(self) -> None: + if self._ensure_task and not self._ensure_task.done(): + self._ensure_task.cancel() + + # -- client ---------------------------------------------------------------- + + @typechecked + async def get_providers(self) -> list[dict] | dict: + return await _get_providers() + + @typechecked + async def get_models(self) -> list[dict]: + return await _get_models() + + @typechecked + async def start_oauth(self, provider: str) -> dict: + return await _start_oauth(provider) + + @typechecked + async def poll_oauth( + self, + provider: str, + device_code: str, + code_verifier: str | None = None, + extra_data: dict | None = None, + ) -> dict: + return await _poll_oauth(provider, device_code, code_verifier=code_verifier, extra_data=extra_data) + + @typechecked + async def exchange_oauth( + self, + provider: str, + code: str, + redirect_uri: str, + code_verifier: str, + state: str = "", + ) -> dict: + return await _exchange_oauth(provider, code, redirect_uri, code_verifier, state) + + @typechecked + async def disconnect_provider(self, provider_id: str) -> bool: + return await _disconnect_provider(provider_id) diff --git a/backend/apps/subscriptions/NineRouter/NineRouterClient.py b/backend/apps/subscriptions/NineRouter/NineRouterClient.py index bf7ccca2..416c6594 100644 --- a/backend/apps/subscriptions/NineRouter/NineRouterClient.py +++ b/backend/apps/subscriptions/NineRouter/NineRouterClient.py @@ -118,3 +118,10 @@ async def get_models() -> list[dict]: except Exception as e: print(f"9Router models fetch failed: {e}") return [] + + +@typechecked +async def disconnect_provider(provider_id: str) -> bool: + async with httpx.AsyncClient(timeout=10.0) as client: + r = await client.delete(f"{NINE_ROUTER_API}/providers/{provider_id}") + return r.status_code == 200 diff --git a/backend/apps/subscriptions/subscriptions.py b/backend/apps/subscriptions/subscriptions.py index 48524c83..29a35009 100644 --- a/backend/apps/subscriptions/subscriptions.py +++ b/backend/apps/subscriptions/subscriptions.py @@ -5,28 +5,17 @@ for the frontend to connect/disconnect subscription providers (Claude, ChatGPT, Gemini, etc.) via OAuth. """ -import asyncio - -import httpx from contextlib import asynccontextmanager -from typing import Optional +from typing import Dict, Optional from fastapi import HTTPException, Request from fastapi.responses import HTMLResponse, JSONResponse from backend.config.Apps import SubApp -from backend.apps.subscriptions.NineRouter.NineRouterProcess.NineRouterProcess import ( - is_running, ensure_running, stop, -) -from backend.apps.subscriptions.NineRouter.NineRouterClient import ( - get_providers, get_models, start_oauth, poll_oauth, exchange_oauth, - NINE_ROUTER_API, -) +from backend.apps.subscriptions.NineRouter.NineRouter import NineRouter from backend.apps.subscriptions.html_constants import SUCCESS_HTML, ERROR_STYLE -from typing import Dict P_PENDING_OAUTH: Dict[str, dict] = {} -P_ENSURE_TASK: Optional[asyncio.Task] = None # --------------------------------------------------------------------------- # Lifespan @@ -34,16 +23,15 @@ P_ENSURE_TASK: Optional[asyncio.Task] = None @asynccontextmanager async def subscriptions_lifespan(): + router: NineRouter = NineRouter.get() try: - await ensure_running() + await router.ensure_running() except Exception as e: print(f"9Router auto-start failed: {e}") yield - global P_ENSURE_TASK - if P_ENSURE_TASK and not P_ENSURE_TASK.done(): - P_ENSURE_TASK.cancel() + router.cancel_ensure_task() try: - stop() + router.stop() except Exception: pass @@ -57,13 +45,12 @@ subscriptions = SubApp("subscriptions", subscriptions_lifespan) @subscriptions.router.get("/status") async def subscriptions_status() -> dict: - global P_ENSURE_TASK - if not is_running(): - if P_ENSURE_TASK is None or P_ENSURE_TASK.done(): - P_ENSURE_TASK = asyncio.create_task(ensure_running()) + router: NineRouter = NineRouter.get() + if not router.is_running(): + await router.ensure_running_background() return {"running": False, "providers": [], "models": []} - providers = await get_providers() - models = await get_models() + providers = await router.get_providers() + models = await router.get_models() return {"running": True, "providers": providers, "models": models} @@ -73,13 +60,14 @@ async def subscriptions_connect(body: dict) -> dict: if not provider: raise HTTPException(status_code=400, detail="provider required") - if not is_running(): - await ensure_running() - if not is_running(): + router: NineRouter = NineRouter.get() + if not router.is_running(): + await router.ensure_running() + if not router.is_running(): raise HTTPException(status_code=503, detail="9Router not available. Please install Node.js.") try: - result: dict = await start_oauth(provider) + result: dict = await router.start_oauth(provider) if result.get("flow") == "authorization_code" and result.get("state"): P_PENDING_OAUTH[result["state"]] = { "provider": provider, @@ -99,7 +87,7 @@ async def subscriptions_poll(body: dict) -> dict: raise HTTPException(status_code=400, detail="provider and device_code required") try: - result: dict = await poll_oauth( + result: dict = await NineRouter.get().poll_oauth( provider, device_code, code_verifier=body.get("code_verifier"), extra_data=body.get("extra_data"), @@ -115,13 +103,13 @@ async def subscriptions_disconnect(body: dict) -> dict: if not provider: raise HTTPException(status_code=400, detail="provider required") + router: NineRouter = NineRouter.get() try: - providers_data = await get_providers() + providers_data = await router.get_providers() connections: list = providers_data.get("connections", []) if isinstance(providers_data, dict) else [] conn = next((c for c in connections if c.get("provider") == provider), None) if conn and conn.get("id"): - async with httpx.AsyncClient(timeout=10.0) as client: - await client.delete(f"{NINE_ROUTER_API}/providers/{conn['id']}") + await router.disconnect_provider(conn["id"]) return {"ok": True} return {"ok": False, "error": "Connection not found"} except Exception as e: @@ -164,7 +152,7 @@ async def subscriptions_callback(request: Request): ) try: - await exchange_oauth( + await NineRouter.get().exchange_oauth( pending["provider"], code, pending["redirect_uri"], pending["code_verifier"], state, )