""" NodeFactory class for creating nodes from JSON configuration. """ from typing import Dict, Any, Type, Optional from api.feature.automationV2.src.models.base_node import BaseNode class NodeRegistrationError(Exception): """Exception raised when node registration fails.""" pass class NodeCreationError(Exception): """Exception raised when node creation fails.""" pass class NodeFactory: """ Factory class for creating node instances from JSON configuration. Provides node type registration and validation mechanisms to support extensible node creation from workflow definitions. """ def __init__(self): """Initialize the node factory with empty registry.""" self._node_registry: Dict[str, Type[BaseNode]] = {} def register_node_type(self, node_type: str, node_class: Type[BaseNode]) -> None: """ Register a node type with its corresponding class. Args: node_type: String identifier for the node type node_class: Class that implements BaseNode interface Raises: NodeRegistrationError: If node_type is invalid or node_class doesn't inherit from BaseNode """ if not node_type or not isinstance(node_type, str): raise NodeRegistrationError("Node type must be a non-empty string") if not issubclass(node_class, BaseNode): raise NodeRegistrationError(f"Node class {node_class.__name__} must inherit from BaseNode") if node_type in self._node_registry: raise NodeRegistrationError(f"Node type '{node_type}' is already registered") self._node_registry[node_type] = node_class def unregister_node_type(self, node_type: str) -> None: """ Unregister a node type. Args: node_type: String identifier for the node type to remove """ if node_type in self._node_registry: del self._node_registry[node_type] def is_registered(self, node_type: str) -> bool: """ Check if a node type is registered. Args: node_type: String identifier for the node type Returns: True if node type is registered, False otherwise """ return node_type in self._node_registry def get_registered_types(self) -> list: """ Get list of all registered node types. Returns: List of registered node type strings """ return list(self._node_registry.keys()) def create_node(self, node_definition: Dict[str, Any], property_id: str = None, database_logger=None) -> BaseNode: """ Create a node instance from JSON configuration. Args: node_definition: Dictionary containing node configuration with required keys: - node_id: Unique identifier for the node - name: Human-readable name for the node - type: Node type string (must be registered) - config: Node-specific configuration dictionary - position: Optional position dictionary with x, y coordinates property_id: Property identifier for the node (optional, can be in node_definition) Returns: BaseNode instance of the appropriate type Raises: NodeCreationError: If node creation fails due to invalid configuration """ # Validate required fields required_fields = ['node_id', 'name', 'type', 'config'] for field in required_fields: if field not in node_definition: raise NodeCreationError(f"Missing required field: {field}") node_id = node_definition['node_id'] name = node_definition['name'] node_type = node_definition['type'] config = node_definition['config'] # Get property_id from parameter or node_definition if property_id is None: property_id = node_definition.get('property_id', 'default') # Validate field types if not isinstance(node_id, str) or not node_id.strip(): raise NodeCreationError("node_id must be a non-empty string") if not isinstance(name, str) or not name.strip(): raise NodeCreationError("name must be a non-empty string") if not isinstance(node_type, str) or not node_type.strip(): raise NodeCreationError("type must be a non-empty string") if not isinstance(config, dict): if not isinstance(config, list): raise NodeCreationError("config must be a dictionary or list") # Check if node type is registered if not self.is_registered(node_type): raise NodeCreationError(f"Unknown node type: {node_type}") # Get node class and create instance node_class = self._node_registry[node_type] try: node = node_class(property_id, node_id, name, config, database_logger) node.node_type = node_type # Set position if provided if 'position' in node_definition: position = node_definition['position'] if isinstance(position, dict) and 'x' in position and 'y' in position: node.set_position(position['x'], position['y']) # Validate the created node's configuration if not node.validate_config(): raise NodeCreationError(f"Node configuration validation failed for {node_id}") return node except Exception as e: if isinstance(e, NodeCreationError): raise raise NodeCreationError(f"Failed to create node {node_id}: {str(e)}") def create_nodes_from_workflow(self, workflow_definition: Dict[str, Any], database_logger=None) -> Dict[str, BaseNode]: """ Create all nodes from a workflow definition. Args: workflow_definition: Dictionary containing workflow configuration with 'nodes' key Returns: Dictionary mapping node_id to BaseNode instances Raises: NodeCreationError: If any node creation fails """ if 'nodes' not in workflow_definition: raise NodeCreationError("Workflow definition must contain 'nodes' key") nodes_config = workflow_definition['nodes'] if not isinstance(nodes_config, dict): raise NodeCreationError("Workflow 'nodes' must be a dictionary") # Get property_id from workflow definition property_id = workflow_definition.get('property_id', 'default') created_nodes = {} for node_id, node_config in nodes_config.items(): # Add node_id to the configuration if not present if 'node_id' not in node_config: node_config['node_id'] = node_id try: node = self.create_node(node_config, property_id, database_logger) created_nodes[node_id] = node except NodeCreationError as e: raise NodeCreationError(f"Failed to create node '{node_id}': {str(e)}") return created_nodes def validate_workflow_nodes(self, workflow_definition: Dict[str, Any]) -> bool: """ Validate all nodes in a workflow definition without creating them. Args: workflow_definition: Dictionary containing workflow configuration Returns: True if all nodes are valid, False otherwise """ try: self.create_nodes_from_workflow(workflow_definition) return True except NodeCreationError: return False def get_node_schema(self, node_type: str) -> Optional[Dict[str, Any]]: """ Get schema information for a registered node type. Args: node_type: String identifier for the node type Returns: Dictionary with node schema information or None if not registered """ if not self.is_registered(node_type): return None node_class = self._node_registry[node_type] # Create a temporary instance to get required config keys try: temp_node = node_class("temp", "temp", "temp", {}) required_keys = temp_node.get_required_config_keys() except Exception: required_keys = [] return { 'node_type': node_type, 'class_name': node_class.__name__, 'required_config_keys': required_keys, 'description': node_class.__doc__ or "No description available" } def clear_registry(self) -> None: """Clear all registered node types.""" self._node_registry.clear() def __len__(self) -> int: """Return number of registered node types.""" return len(self._node_registry) def __contains__(self, node_type: str) -> bool: """Check if node type is registered using 'in' operator.""" return self.is_registered(node_type) def __str__(self) -> str: """String representation of the factory.""" return f"NodeFactory({len(self._node_registry)} registered types)" def __repr__(self) -> str: """Detailed string representation of the factory.""" types = list(self._node_registry.keys()) return f"NodeFactory(types={types})"