from opensearchpy import OpenSearch import json import urllib3 import requests import os from .base_db import BaseDBHandler # Disable SSL warnings for self-signed certificates urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) class OpenSearchHandler(BaseDBHandler): def connect(self, service, user, password, database=None): if ':12000' in service: base_url = service else: base_url = f"{service}:12000" auth = None if user and password: auth = (user, password) # Make the health check request response = requests.get( base_url, auth=auth, timeout=20, verify=False ) if response.status_code == 200: self.base_url = base_url self.auth = auth basic_data = response.json() cluster_name = basic_data.get('cluster_name', 'Unknown') # Try to get detailed health status try: health_url = f"{base_url}/_cluster/health" health_response = requests.get(health_url, auth=auth, timeout=10, verify=False) if health_response.status_code == 200: health_data = health_response.json() status = health_data.get('status', 'unknown') # green/yellow/red nodes = health_data.get('number_of_nodes', 'unknown') return self.make_connection_response( f"Successfully connected to OpenSearch: {cluster_name} " f"(/_cluster/health -> status: {status}, number of nodes: {nodes})" ) else: # Health endpoint not available, return only basic info return self.make_connection_response(f"Successfully connected to OpenSearch: {cluster_name}, But /_cluster/health is not available.") except: return self.make_connection_response(f"Successfully connected to OpenSearch: {cluster_name}, But Heath check was failed.") else: return self.make_error_response(response.status_code, f"OpenSearch connection failed: HTTP {response.status_code}") def execute_query(self, query, database=None): try: # Try to parse as JSON try: query_json = json.loads(query) except json.JSONDecodeError: return self.make_error_response(400, "Invalid JSON query. OpenSearch requires JSON format queries.") # Basic search query handling if 'query' in query_json: index = query_json.get('index', '_all') search_body = {'query': query_json['query']} if 'size' in query_json: search_body['size'] = query_json['size'] result = self.client.search(index=index, body=search_body) hits = result.get('hits', {}).get('hits', []) total = result.get('hits', {}).get('total', {}) if isinstance(total, dict): total_value = total.get('value', 0) else: total_value = total if hits: # Extract field names from first hit first_hit = hits[0] source_keys = list(first_hit.get('_source', {}).keys()) columns_names = ['_index', '_id', '_score'] + source_keys results = [] for hit in hits: row = [ hit.get('_index', ''), hit.get('_id', ''), hit.get('_score', 0) ] source = hit.get('_source', {}) for key in source_keys: row.append(source.get(key, None)) results.append(row) return self.make_success_response( msg=f"Found {total_value} results", columns_names=columns_names, results=results ) else: return self.make_success_response( msg=f"Search completed, {total_value} results found", columns_names=[], results=[] ) else: return self.make_error_response(400, "Unsupported query format. Use the OpenSearch API interface for full functionality.") except Exception as e: return self.make_error_response(500, f"Internal Server Error, Please check the query and try again: {e}") def execute_api_query(self, method, endpoint, payload=None): """Execute OpenSearch API query - this is the main method for API calls""" try: # Clean up endpoint if not endpoint.startswith('/'): endpoint = '/' + endpoint full_url = f"{self.base_url}{endpoint}" # Execute API call based on method if method.upper() == 'GET': if payload: try: payload_json = json.loads(payload) if isinstance(payload, str) else payload response = requests.get( full_url, auth=self.auth, json=payload_json, verify=False, timeout=20 ) except json.JSONDecodeError: return self.make_error_response(400, "Invalid JSON payload") else: response = requests.get( full_url, auth=self.auth, verify=False, timeout=20 ) elif method.upper() == 'POST': if payload: try: payload_json = json.loads(payload) if isinstance(payload, str) else payload response = requests.post( full_url, auth=self.auth, json=payload_json, verify=False, timeout=20 ) except json.JSONDecodeError: return self.make_error_response(400, "Invalid JSON payload") else: response = requests.post( full_url, auth=self.auth, verify=False, timeout=20 ) elif method.upper() == 'PUT': if payload: try: payload_json = json.loads(payload) if isinstance(payload, str) else payload response = requests.put( full_url, auth=self.auth, json=payload_json, verify=False, timeout=20 ) except json.JSONDecodeError: return self.make_error_response(400, "Invalid JSON payload") else: response = requests.put( full_url, auth=self.auth, verify=False, timeout=20 ) elif method.upper() == 'DELETE': response = requests.delete( full_url, auth=self.auth, verify=False, timeout=20 ) else: return self.make_error_response(400, f"Unsupported HTTP method: {method}") # Handle response if response.status_code >= 200 and response.status_code < 300: try: result = response.json() except: result = response.text return { "code": response.status_code, "msg": f"{method.upper()} {endpoint} executed successfully", "response": result } else: return self.make_error_response(response.status_code, f"API call failed: HTTP {response.status_code}") except Exception as e: return self.make_error_response(500, f"Internal Server Error, API call failed: {e}")