| import logging |
| import httpx |
| import os |
| import asyncio |
| from typing import Dict, Any, List, Optional |
| from datetime import datetime, timezone, timedelta |
|
|
| logger = logging.getLogger(__name__) |
|
|
| class NodeBridgeService: |
| def __init__(self): |
| self.node_url = os.getenv("NODE_ENGINE_URL", "http://localhost:3003") |
| self.client = httpx.AsyncClient(base_url=self.node_url, timeout=30.0) |
| |
| |
| self._catalog_cache: List[Dict[str, Any]] = [] |
| self._catalog_last_updated: datetime = datetime.min |
| self._cache_ttl = timedelta(minutes=5) |
|
|
| async def get_health(self) -> bool: |
| try: |
| resp = await self.client.get("/health") |
| resp.raise_for_status() |
| return True |
| except Exception as e: |
| logger.error(f"Node engine health check failed: {e}") |
| return False |
|
|
| async def get_catalog(self, force_refresh: bool = False) -> List[Dict[str, Any]]: |
| """Fetches pieces from Node engine.""" |
| if not force_refresh and self._catalog_cache and (datetime.now(timezone.utc) - self._catalog_last_updated < self._cache_ttl): |
| return self._catalog_cache |
|
|
| try: |
| resp = await self.client.get("/pieces") |
| resp.raise_for_status() |
| pieces = resp.json() |
|
|
| self._catalog_cache = pieces |
| self._catalog_last_updated = datetime.now(timezone.utc) |
| return pieces |
| except Exception as e: |
| logger.error(f"Failed to fetch catalog from Node engine: {e}") |
| return [] |
|
|
| async def get_piece_details(self, piece_name: str) -> Optional[Dict[str, Any]]: |
| """Get piece details.""" |
| try: |
| resp = await self.client.get(f"/pieces/{piece_name}") |
| resp.raise_for_status() |
| return resp.json() |
| except httpx.HTTPStatusError as e: |
| if e.response.status_code == 404: |
| return None |
| logger.error(f"Failed to get piece details for {piece_name}: {e}") |
| return None |
| except Exception as e: |
| return None |
|
|
| async def execute_action(self, |
| piece_name: str, |
| action_name: str, |
| props: Dict[str, Any], |
| auth: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: |
| """Executes a piece action.""" |
| payload = { |
| "pieceName": piece_name, |
| "actionName": action_name, |
| "props": props, |
| "auth": auth |
| } |
|
|
| try: |
| resp = await self.client.post("/execute/action", json=payload) |
| resp.raise_for_status() |
| result = resp.json() |
|
|
| if not result.get("success", False): |
| raise Exception(f"Execution failed: {result.get('error', 'Unknown error')}") |
|
|
| return result.get("output", {}) |
| except Exception as e: |
| logger.error(f"Execution error for {piece_name}.{action_name}: {e}") |
| raise |
|
|
| async def close(self): |
| await self.client.aclose() |
|
|
| |
| node_bridge = NodeBridgeService() |
|
|
