from neo4j import GraphDatabase, Result from dotenv import load_dotenv import os from typing import Tuple, TypedDict import logging from pathlib import Path from adminable_tenants import format_adminable_tenants_result class QueryResult(TypedDict): records: list[dict[str, any]] summary: dict[str, any] keys: list[str] def analyze_profile(profile_data): def extract_metrics(node, depth=0): metrics = { 'operatorType': node.get('operatorType', ''), 'rows': node.get('rows', 0), 'dbHits': node.get('dbHits', 0), 'time': node.get('time', 0), 'pageCacheHits': node.get('pageCacheHits', 0), 'pageCacheMisses': node.get('pageCacheMisses', 0), 'details': node.get('args', {}).get('Details', ''), 'memory': node.get('args', {}).get('Memory', 0), 'estimatedRows': node.get('args', {}).get('EstimatedRows', 0), 'depth': depth } child_metrics = [] for child in node.get('children', []): child_metrics.extend(extract_metrics(child, depth + 1)) return [metrics] + child_metrics all_metrics = extract_metrics(profile_data) # Calculate totals totals = { 'total_db_hits': sum(m['dbHits'] for m in all_metrics), 'total_rows': sum(m['rows'] for m in all_metrics), 'total_time': sum(m['time'] for m in all_metrics if m['time'] > 0), 'total_cache_hits': sum(m['pageCacheHits'] for m in all_metrics), 'total_cache_misses': sum(m['pageCacheMisses'] for m in all_metrics), 'total_memory': sum(m['memory'] for m in all_metrics) } # Format the output formatted_output = "Query Profile Summary:\n" formatted_output += f"Total DB Hits: {totals['total_db_hits']:,}\n" formatted_output += f"Total Rows Processed: {totals['total_rows']:,}\n" formatted_output += f"Total Time: {totals['total_time']:,}ms\n" formatted_output += f"Total Cache Hits: {totals['total_cache_hits']:,}\n" formatted_output += f"Total Memory Used: {totals['total_memory']:,} bytes\n\n" formatted_output += "Operation Breakdown:\n" for metric in all_metrics: indent = " " * metric['depth'] formatted_output += f"{indent}→ {metric['operatorType']}\n" formatted_output += f"{indent} Rows: {metric['rows']:,}\n" formatted_output += f"{indent} DB Hits: {metric['dbHits']:,}\n" if metric['time'] > 0: formatted_output += f"{indent} Time: {metric['time']:,}ms\n" if metric['details']: formatted_output += f"{indent} Details: {metric['details']}\n" formatted_output += "\n" return formatted_output def _print_summary(result: Result): summary = result.consume() # Timing information print(f"Query time: {summary.result_available_after} ms") print(f"Result consumption time: {summary.result_consumed_after} ms") # Plan information (if available) if summary.plan: print(f"Plan: {summary.plan}") if summary.profile: print(analyze_profile(summary.profile)) def compare_tenant_arrays(array1, array2): """ Compare two arrays of dictionaries containing tenant IDs. Returns sets of unique tenant IDs that are: 1. Present in both arrays 2. Only in first array 3. Only in second array Args: array1: List of dictionaries with 'tenant_id' key array2: List of dictionaries with 'tenant_id' key Returns: tuple: (common_ids, only_in_first, only_in_second) """ # Extract tenant IDs into sets for efficient comparison tenant_ids1 = {item['tenant_id'] for item in array1} tenant_ids2 = {item['tenant_id'] for item in array2} # Find common and unique tenant IDs common_ids = tenant_ids1.intersection(tenant_ids2) only_in_first = tenant_ids1.difference(tenant_ids2) only_in_second = tenant_ids2.difference(tenant_ids1) # Get frequency counts for duplicate analysis freq1 = {} freq2 = {} for item in array1: tenant_id = item['tenant_id'] freq1[tenant_id] = freq1.get(tenant_id, 0) + 1 for item in array2: tenant_id = item['tenant_id'] freq2[tenant_id] = freq2.get(tenant_id, 0) + 1 # Print duplicate analysis print("\nDuplicate Analysis:") for tenant_id in common_ids: if freq1[tenant_id] > 1 or freq2[tenant_id] > 1: print(f"Tenant ID {tenant_id}:") print(f" Occurrences in first array: {freq1[tenant_id]}") print(f" Occurrences in second array: {freq2[tenant_id]}") return common_ids, only_in_first, only_in_second class CypherRunner: """A simple Neo4j query runner that uses environment variables for configuration.""" def __init__(self, queries_dir: str = 'cypher'): load_dotenv() logging.basicConfig(level=logging.INFO) self.logger = logging.getLogger(__name__) self.queries_dir = Path(queries_dir) if not self.queries_dir.exists(): self.queries_dir.mkdir(parents=True) self.logger.info(f"Created queries directory: {self.queries_dir}") self.uri = os.getenv('NEO4J_URL') self.user = os.getenv('NEO4J_USERNAME') self.password = os.getenv('NEO4J_PASSWORD') self.max_retry_time = int(os.getenv('NEO4J_MAX_RETRY_TIME', '30')) if not all([self.uri, self.user, self.password]): raise ValueError("Missing required environment variables. Please check your .env file.") self.driver = GraphDatabase.driver( self.uri, auth=(self.user, self.password), max_transaction_retry_time=self.max_retry_time ) try: self.driver.verify_connectivity() self.logger.info("Successfully connected to Neo4j database") except Exception as e: self.logger.error(f"Failed to connect to Neo4j database: {e}") raise def close(self): """Close the database driver.""" self.driver.close() def load_query_file(self, filename: str) -> str: """ Load a Cypher query from a file in the queries directory. Args: filename: Name of the query file (with or without .cypher extension) Returns: The query string from the file """ if not filename.endswith('.cypher'): filename = f"{filename}.cypher" query_path = self.queries_dir / filename print(query_path) try: with open(query_path, 'r') as f: query = f.read().strip() self.logger.info(f"Successfully loaded query from {filename}") return query except FileNotFoundError: raise FileNotFoundError(f"Query file not found: {filename}") except Exception as e: self.logger.error(f"Error loading query file {filename}: {e}") raise def run_query( self, query: str, parameters: dict[str, any] = None, print_summary: bool = False ) -> list[dict[str, any]]: """ Execute a Cypher query and return the results. Args: query: The Cypher query to execute parameters: Optional dictionary of query parameters Returns: List of dictionaries containing the query results """ parameters = parameters or {} try: with self.driver.session() as session: result = session.run(query, parameters) records = [dict(record) for record in result] if print_summary: _print_summary(result) return records except Exception as e: self.logger.error(f"Query execution failed: {e}") raise def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): self.close() if __name__ == "__main__": identity_id = '47a0f3a9-2a3e-4dd5-9164-56fdba37b982' try: with CypherRunner() as runner: query = runner.load_query_file('adminableTypes.cypher') results = runner.run_query( query, { 'userIdentityId': identity_id } ) adminable_tenants = format_adminable_tenants_result(results) print(len(adminable_tenants)) tenant_dicts = [at['tenant'] for at in adminable_tenants] query = runner.load_query_file('matchWithUnwind.cypher') r1 = runner.run_query( query=query, parameters={ 'identityId': identity_id, 'tenants': tenant_dicts, }, print_summary=True ) print('----------------') query = runner.load_query_file('matchWithoutUnwind.cypher') r2 = runner.run_query( query=query, parameters={ 'identityId': identity_id, 'tenants': tenant_dicts, }, print_summary=True ) except Exception as e: logging.error(f"Error occurred: {e}")