""" Connector Manager - Manages all client connectors Responsibilities: 1. Create/load connectors based on client configuration 2. Cache active connectors for performance 3. Handle connection lifecycle 4. Provide connector instances to the Tool Executor """ from typing import Dict, Optional, Type import logging from datetime import datetime, timedelta from app.connectors.base import ( BaseConnector, ConnectorConfig, ConnectorType, ConnectionStatus, ) from app.connectors.custom_api import CustomAPIConnector from app.connectors.custom_database import CustomDatabaseConnector from app.connectors.platforms.shopify import ShopifyConnector from app.connectors.platforms.woocommerce import WooCommerceConnector logger = logging.getLogger(__name__) # Registry of connector classes CONNECTOR_CLASSES: Dict[ConnectorType, Type[BaseConnector]] = { ConnectorType.CUSTOM_API: CustomAPIConnector, ConnectorType.CUSTOM_DATABASE: CustomDatabaseConnector, ConnectorType.SHOPIFY: ShopifyConnector, ConnectorType.WOOCOMMERCE: WooCommerceConnector, } class CachedConnector: """Wrapper for cached connector with metadata""" def __init__(self, connector: BaseConnector): self.connector = connector self.created_at = datetime.utcnow() self.last_used = datetime.utcnow() self.use_count = 0 def touch(self): """Update last used time""" self.last_used = datetime.utcnow() self.use_count += 1 @property def age_seconds(self) -> float: return (datetime.utcnow() - self.created_at).total_seconds() @property def idle_seconds(self) -> float: return (datetime.utcnow() - self.last_used).total_seconds() class ConnectorManager: """ Central manager for all client connectors. Features: - Lazy loading of connectors - Connection caching - Automatic reconnection - Connection pooling """ # Cache settings MAX_CACHED_CONNECTORS = 100 IDLE_TIMEOUT_SECONDS = 1800 # 30 minutes MAX_AGE_SECONDS = 3600 # 1 hour def __init__(self): self._connectors: Dict[int, CachedConnector] = {} self._configs: Dict[int, ConnectorConfig] = {} async def get_connector(self, client_id: int) -> Optional[BaseConnector]: """ Get a connector for a client. If the connector exists in cache and is healthy, return it. Otherwise, create a new connector. Args: client_id: Client ID Returns: Connector instance or None if not configured """ # Check cache if client_id in self._connectors: cached = self._connectors[client_id] # Check if still valid if self._is_valid(cached): cached.touch() return cached.connector # Remove stale connector await self._remove_connector(client_id) # Create new connector config = await self._get_config(client_id) if not config: logger.warning(f"No connector config for client {client_id}") return None connector = await self._create_connector(client_id, config) if connector: self._connectors[client_id] = CachedConnector(connector) await self._cleanup_cache() return connector return None async def _get_config(self, client_id: int) -> Optional[ConnectorConfig]: """ Get connector configuration for a client. In production, this would fetch from database. """ # Check local cache first if client_id in self._configs: return self._configs[client_id] # TODO: Fetch from database # For now, return None (will use mock connector) # In production: # config_data = await db.fetch_connector_config(client_id) # if config_data: # config = ConnectorConfig(**config_data) # self._configs[client_id] = config # return config return None async def _create_connector( self, client_id: int, config: ConnectorConfig ) -> Optional[BaseConnector]: """Create and connect a new connector""" connector_class = CONNECTOR_CLASSES.get(config.connector_type) if not connector_class: logger.error(f"Unknown connector type: {config.connector_type}") return None try: connector = connector_class(config, client_id) success = await connector.connect() if success: logger.info(f"Created connector for client {client_id}: {config.connector_type}") return connector else: logger.error(f"Failed to connect: {connector.last_error}") return None except Exception as e: logger.exception(f"Error creating connector for client {client_id}: {e}") return None def _is_valid(self, cached: CachedConnector) -> bool: """Check if cached connector is still valid""" # Check if connection is still active if cached.connector.status != ConnectionStatus.CONNECTED: return False # Check idle timeout if cached.idle_seconds > self.IDLE_TIMEOUT_SECONDS: return False # Check max age if cached.age_seconds > self.MAX_AGE_SECONDS: return False return True async def _remove_connector(self, client_id: int): """Remove and disconnect a connector""" if client_id in self._connectors: cached = self._connectors[client_id] try: await cached.connector.disconnect() except Exception as e: logger.warning(f"Error disconnecting connector: {e}") del self._connectors[client_id] async def _cleanup_cache(self): """Clean up old connectors to prevent memory issues""" # Remove if over limit while len(self._connectors) > self.MAX_CACHED_CONNECTORS: # Find oldest idle connector oldest_id = min( self._connectors.keys(), key=lambda k: self._connectors[k].last_used ) await self._remove_connector(oldest_id) # Remove stale connectors stale_ids = [ cid for cid, cached in self._connectors.items() if not self._is_valid(cached) ] for cid in stale_ids: await self._remove_connector(cid) # ==================== CONFIGURATION METHODS ==================== async def register_connector( self, client_id: int, config: ConnectorConfig ) -> Dict: """ Register a new connector configuration for a client. Args: client_id: Client ID config: Connector configuration Returns: Result with success status and test results """ # Validate config if config.connector_type not in CONNECTOR_CLASSES: return { "success": False, "error": f"Unsupported connector type: {config.connector_type}" } # Test the connection connector = await self._create_connector(client_id, config) if not connector: return { "success": False, "error": "Failed to establish connection" } test_result = await connector.test_connection() if test_result["success"]: # Store config self._configs[client_id] = config self._connectors[client_id] = CachedConnector(connector) # TODO: Save to database # await db.save_connector_config(client_id, config.dict()) return { "success": True, "message": "Connector configured successfully", "test_result": test_result } else: await connector.disconnect() return { "success": False, "error": test_result["message"], "details": test_result.get("details") } async def update_connector( self, client_id: int, config: ConnectorConfig ) -> Dict: """Update connector configuration""" # Remove existing connector await self._remove_connector(client_id) # Register new config return await self.register_connector(client_id, config) async def remove_connector(self, client_id: int) -> Dict: """Remove connector configuration""" if client_id in self._connectors: await self._remove_connector(client_id) if client_id in self._configs: del self._configs[client_id] # TODO: Remove from database # await db.delete_connector_config(client_id) return {"success": True, "message": "Connector removed"} async def test_connector(self, client_id: int) -> Dict: """Test an existing connector""" connector = await self.get_connector(client_id) if not connector: return { "success": False, "error": "Connector not found or not configured" } return await connector.test_connection() async def get_connector_status(self, client_id: int) -> Dict: """Get status of a client's connector""" if client_id not in self._connectors: if client_id in self._configs: return { "configured": True, "connected": False, "status": "not_loaded" } return { "configured": False, "connected": False, "status": "not_configured" } cached = self._connectors[client_id] return { "configured": True, "connected": cached.connector.status == ConnectionStatus.CONNECTED, "status": cached.connector.status.value, "connector_type": cached.connector.connector_type.value, "use_count": cached.use_count, "idle_seconds": int(cached.idle_seconds), "age_seconds": int(cached.age_seconds) } async def list_connectors(self) -> Dict: """List all configured connectors""" connectors = [] for client_id, cached in self._connectors.items(): connectors.append({ "client_id": client_id, "type": cached.connector.connector_type.value, "name": cached.connector.config.name, "status": cached.connector.status.value, "use_count": cached.use_count }) return { "count": len(connectors), "connectors": connectors } # ==================== LIFECYCLE METHODS ==================== async def shutdown(self): """Shutdown all connectors gracefully""" logger.info("Shutting down connector manager...") for client_id in list(self._connectors.keys()): await self._remove_connector(client_id) logger.info("All connectors disconnected") # Global singleton instance connector_manager = ConnectorManager()