""" API Call Node for external service integrations. """ import re import json import logging from datetime import datetime from typing import Dict, Any, Optional, Union, List from urllib.parse import urljoin from ..models.base_node import BaseNode from ..models.execution_context import ExecutionContext from ..models.execution_result import ExecutionResult from ..integrations.http_client import HTTPClient, HTTPClientError logger = logging.getLogger(__name__) class APICallNode(BaseNode): """ API Call Node for making HTTP requests to external services. Supports all HTTP methods (GET, POST, PUT, DELETE) with context variable substitution, authentication, retry logic, and parallel execution capabilities. """ def __init__(self, property_id: str, node_id: str, name: str, config: Dict[str, Any], database_logger=None): """ Initialize API Call Node. Args: property_id: Property identifier for this node instance node_id: Unique identifier for this node instance name: Human-readable name for the node config: Node configuration containing HTTP request details database_logger: Optional database logger for node-level logging """ super().__init__(property_id, node_id, name, config, database_logger) self.node_type = "api_call" self._http_client: Optional[HTTPClient] = None def get_required_config_keys(self) -> List[str]: """ Get list of required configuration keys. Returns: List of required configuration keys """ return ["method", "url"] def validate_config(self) -> bool: """ Validate the node's configuration. Returns: True if configuration is valid, False otherwise """ if not self.validate_required_config(): return False # Validate HTTP method method = self.config.get("method", "").upper() if method not in ["GET", "POST", "PUT", "DELETE", "PATCH", "HEAD", "OPTIONS"]: logger.error(f"Invalid HTTP method: {method}") return False # Validate URL url = self.config.get("url", "") if not url or not isinstance(url, str): logger.error("URL must be a non-empty string") return False # Validate retry configuration if present retry_config = self.config.get("retry", {}) if retry_config: if not isinstance(retry_config, dict): logger.error("Retry configuration must be a dictionary") return False attempts = retry_config.get("attempts", 3) if not isinstance(attempts, int) or attempts < 0: logger.error("Retry attempts must be a non-negative integer") return False delay = retry_config.get("delay", 1000) if not isinstance(delay, (int, float)) or delay < 0: logger.error("Retry delay must be a non-negative number") return False return True def _get_http_client(self) -> HTTPClient: """ Get or create HTTP client instance. Returns: HTTPClient instance """ if self._http_client is None: timeout = self.config.get("timeout", 30) default_headers = self.config.get("default_headers", {}) self._http_client = HTTPClient(timeout=timeout, default_headers=default_headers) return self._http_client def _substitute_context_variables(self, text: str, context: ExecutionContext) -> str: """ Substitute context variables in text using {{context.key}} syntax. Args: text: Text containing variable placeholders context: Execution context with data and variables Returns: Text with variables substituted """ if not isinstance(text, str): return text # Pattern to match {{context.key}} or {{variables.key}} pattern = r'\{\{(context|variables)\.([^}]+)\}\}' def replace_variable(match): source = match.group(1) # 'context' or 'variables' key = match.group(2) # variable key try: if source == "context": value = context.get_data(key) else: # variables value = context.get_variable(key) if value is None: logger.warning(f"Variable {source}.{key} not found in context") return match.group(0) # Return original placeholder # Convert to string if not already if isinstance(value, (dict, list)): return json.dumps(value) return str(value) except Exception as e: logger.error(f"Error substituting variable {source}.{key}: {e}") return match.group(0) # Return original placeholder return re.sub(pattern, replace_variable, text) def _substitute_context_in_dict(self, data: Dict[str, Any], context: ExecutionContext) -> Dict[str, Any]: """ Recursively substitute context variables in dictionary values. Args: data: Dictionary containing potential variable placeholders context: Execution context with data and variables Returns: Dictionary with variables substituted """ result = {} for key, value in data.items(): if isinstance(value, str): result[key] = self._substitute_context_variables(value, context) elif isinstance(value, dict): result[key] = self._substitute_context_in_dict(value, context) elif isinstance(value, list): result[key] = [ self._substitute_context_variables(item, context) if isinstance(item, str) else self._substitute_context_in_dict(item, context) if isinstance(item, dict) else item for item in value ] else: result[key] = value return result def _prepare_request_data(self, context: ExecutionContext) -> Dict[str, Any]: """ Prepare request data with context variable substitution. Args: context: Execution context Returns: Dictionary containing prepared request parameters """ # Get base configuration method = self.config.get("method", "GET").upper() url = self._substitute_context_variables(self.config.get("url", ""), context) # Prepare headers headers = self.config.get("headers", {}) if headers: headers = self._substitute_context_in_dict(headers, context) # Prepare query parameters params = self.config.get("params", {}) if params: params = self._substitute_context_in_dict(params, context) # Prepare request body body = self.config.get("body") json_data = None form_data = None if body: if isinstance(body, dict): # Check if we should send as JSON or form data content_type = headers.get("Content-Type", "").lower() if "application/json" in content_type or not content_type: json_data = self._substitute_context_in_dict(body, context) else: form_data = self._substitute_context_in_dict(body, context) elif isinstance(body, str): form_data = self._substitute_context_variables(body, context) # Prepare authentication auth = None auth_config = self.config.get("auth") if auth_config and isinstance(auth_config, dict): auth_type = auth_config.get("type", "").lower() if auth_type == "basic": username = self._substitute_context_variables(auth_config.get("username", ""), context) password = self._substitute_context_variables(auth_config.get("password", ""), context) auth = (username, password) # Prepare retry configuration retry_config = self.config.get("retry", {}) retry_attempts = retry_config.get("attempts", 3) retry_delay = retry_config.get("delay", 1000) / 1000.0 # Convert ms to seconds return { "method": method, "url": url, "headers": headers, "params": params, "json_data": json_data, "form_data": form_data, "auth": auth, "retry_attempts": retry_attempts, "retry_delay": retry_delay } def _execute_http_request(self, request_data: Dict[str, Any], user_pseudo_id:str = None) -> Dict[str, Any]: """ Execute HTTP request using the HTTP client. Args: request_data: Prepared request data Returns: Dictionary containing response data Raises: HTTPClientError: If request fails """ client = self._get_http_client() method = request_data["method"] # Prepare request arguments request_args = { "url": request_data["url"], "headers": request_data["headers"], "auth": request_data["auth"], "retry_attempts": request_data["retry_attempts"], "retry_delay": request_data["retry_delay"] } # Add method-specific arguments if method == "GET": request_args["params"] = request_data["params"] response = client.get(**request_args) elif method in ["POST", "PUT", "PATCH"]: request_args["params"] = request_data["params"] if request_data["json_data"]: request_args["json_data"] = request_data["json_data"] request_args["json_data"]['user_pseudo_id'] = user_pseudo_id elif request_data["form_data"]: request_args["data"] = request_data["form_data"] request_args["data"]['user_pseudo_id'] = user_pseudo_id if method == "POST": response = client.post(**request_args) elif method == "PUT": response = client.put(**request_args) else: # PATCH - use requests directly through session response = client.session.request(method, **request_args) elif method == "DELETE": response = client.delete(**request_args) else: # For other methods, use session directly response = client.session.request(method, **request_args) # Process response response_data = { "status_code": response.status_code, "headers": dict(response.headers), "url": response.url } # Try to parse JSON response try: if response.content: response_data["data"] = response.json() else: response_data["data"] = None except (ValueError, json.JSONDecodeError): # If not JSON, store as text response_data["data"] = response.text return response_data def execute(self, context: ExecutionContext) -> ExecutionResult: """ Execute the API call node. Args: context: Current execution context Returns: ExecutionResult with API response data or error information """ start_time = datetime.utcnow() # print(context.data[str(context.previous_node[0])]['user_pseudo_ids']) # self.user_pseudo_ids = context.data[str(context.previous_node[0])]['user_pseudo_ids'] if 'user_pseudo_ids' in context.data[str(context.previous_node[0])] else None if not context.data["user_pseudo_ids"]: logger.error(f"No audience in node {self.node_id}") result = ExecutionResult.error_result( error_message=f"No audience in node", node_id=self.node_id, data={ f"{self.node_id}": { "user_pseudo_ids": [], "log_result": [] } } ) result.set_execution_time(start_time) return result try: print(f"Executing API call node {self.node_id}: {self.name}") # Validate configuration if not self.validate_config(): return ExecutionResult.error_result( error_message="Invalid node configuration", node_id=self.node_id ) # Prepare request data with context substitution request_data = self._prepare_request_data(context) print(f"Making {request_data['method']} request to {request_data['url']}") # Execute HTTP request log_result = [] user_pseudo_ids_success = [] for user in context.data["user_pseudo_ids"]: response_data = self._execute_http_request(request_data, user) data = { # "api_response": response_data, "status": 'completed' if response_data['status_code'] == 200 else 'failed', "message": response_data['data'], "request_method": request_data["method"], "request_url": request_data["url"], "user_pseudo_id": user } log_result.append(data) if response_data['status_code'] == 200: user_pseudo_ids_success.append(user) # # Log to database if logger is available # if self.database_logger: # try: # self.database_logger.log_user_pseudo_id( # workflow_id=context.data.get('workflow_id'), # execution_id=context.data.get('execution_id'), # node_id=self.node_id, # user_pseudo_id=user, # status="completed" if response_data['status_code'] == 200 else "failed" # ) # except Exception as e: # logger.warning(f"Failed to log user pseudo id: {e}") # Create success result result = ExecutionResult.success_result( data={ f"{self.node_id}": { "user_pseudo_ids": user_pseudo_ids_success, "log_result": log_result } }, node_id=self.node_id ) result.set_execution_time(start_time) logger.info(f"API call completed successfully: {result}") return result except HTTPClientError as e: logger.error(f"HTTP client error in node {self.node_id}: {e}") result = ExecutionResult.error_result( error_message=f"HTTP request failed: {str(e)}", node_id=self.node_id ) result.set_execution_time(start_time) return result except Exception as e: logger.error(f"Unexpected error in API call node {self.node_id}: {e}") result = ExecutionResult.error_result( error_message=f"Unexpected error: {str(e)}", node_id=self.node_id ) result.set_execution_time(start_time) return result def supports_parallel_execution(self) -> bool: """ Check if this node supports parallel execution. Returns: True, as API calls can be executed in parallel """ return True def get_parallel_execution_key(self) -> str: """ Get key for grouping parallel executions. Returns: Key for parallel execution grouping """ return self.config.get("parallel_group", f"api_call_{self.node_id}") def cleanup(self): """Clean up resources.""" if self._http_client: self._http_client.close() self._http_client = None def __del__(self): """Destructor to ensure cleanup.""" self.cleanup()