🔍 Code Extractor

Executes a custom Cypher query against a Neo4j database and returns the results as a list of dictionaries, handling conversion of Neo4j-specific types.

Purpose

This function provides a flexible interface for running arbitrary Cypher queries against a Neo4j graph database. It handles database connection management, parameter binding, result conversion from Neo4j Record objects to Python dictionaries, and graceful error handling. Use this when you need to execute custom queries that aren't covered by higher-level database abstraction methods.

Source Code

def run_query(query: str, params: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
    """
    Run a custom Cypher query.
    
    Args:
        query: Cypher query string
        params: Query parameters
        
    Returns:
        List of records as dictionaries
    """
    try:
        if params is None:
            params = {}
            
        driver = get_driver()
        with driver.session() as session:
            result = session.run(query, **params)
            
            records = []
            for record in result:
                # Convert Record to dictionary
                record_dict = {}
                for key in record.keys():
                    value = record[key]
                    # Convert Neo4j Node and Relationship types to dictionaries
                    if hasattr(value, 'items'):
                        record_dict[key] = dict(value.items())
                    else:
                        record_dict[key] = value
                        
                records.append(record_dict)
                
            return records
            
    except Exception as e:
        logger.error(f"Error running query: {e}")
        return []

Parameters

Parameter Details

query: A Cypher query string to execute against the Neo4j database. Should be a valid Cypher statement (e.g., 'MATCH (n:Person) RETURN n' or 'CREATE (n:Node {name: $name})'). Can include parameter placeholders using $ syntax.

params: Optional dictionary of parameters to bind to the query. Keys should match parameter names in the query (without the $ prefix). Values can be any JSON-serializable type. Defaults to an empty dictionary if not provided. Example: {'name': 'John', 'age': 30}

Return Value

Type: List[Dict[str, Any]]

Returns a list of dictionaries where each dictionary represents one record from the query result. Dictionary keys correspond to the column names in the RETURN clause of the Cypher query. Neo4j Node and Relationship objects are automatically converted to dictionaries containing their properties. Returns an empty list if the query fails or produces no results.

Dependencies

• neo4j
• logging
• typing

Required Imports

from typing import Dict, List, Any, Optional
import logging
from neo4j import Driver
from CDocs.db import get_driver

Usage Example

# Simple query without parameters
results = run_query('MATCH (n:Person) RETURN n.name as name, n.age as age')
for record in results:
    print(f"Name: {record['name']}, Age: {record['age']}")

# Query with parameters
query = 'MATCH (n:Person {name: $name}) RETURN n'
params = {'name': 'John Doe'}
results = run_query(query, params)

# Create node with parameters
create_query = 'CREATE (n:Person {name: $name, age: $age}) RETURN n'
params = {'name': 'Jane Smith', 'age': 28}
results = run_query(create_query, params)

# Complex query with relationships
query = '''
MATCH (p:Person)-[r:KNOWS]->(friend:Person)
WHERE p.name = $name
RETURN p, r, friend
'''
params = {'name': 'Alice'}
results = run_query(query, params)

Best Practices

• Always use parameterized queries (params argument) instead of string concatenation to prevent Cypher injection attacks
• Handle the possibility of an empty list return value, which indicates either no results or an error
• Check logs for error details when an empty list is returned unexpectedly
• Be aware that Neo4j Node and Relationship objects are converted to dictionaries, losing type information
• Ensure the get_driver() function is properly configured before calling this function
• Use appropriate Cypher query syntax for Neo4j version in use
• Consider query performance and use LIMIT clauses for potentially large result sets
• The function uses a session context manager which automatically closes the session after execution

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function run_query_v1 83.1% similar

  Executes a Cypher query against a Neo4j database session and returns the result, with optional parameterization for safe query execution.

  From: /tf/active/vicechatdev/offline_docstore_multi_vice.py

• function run_query_v2 79.7% similar

  Executes a Cypher query against a Neo4j graph database and returns the results as a list of dictionaries.

  From: /tf/active/vicechatdev/neo4j_schema/neo4j_python_snippets.py

• function evaluate_query 71.7% similar

  Executes a Cypher query against a Neo4j database session and returns the first value from a single result record.

  From: /tf/active/vicechatdev/offline_docstore_multi_vice.py

• function search_documents 53.7% similar

  Searches for documents in a Neo4j graph database based on multiple optional filter criteria including text query, document type, department, status, and owner.

  From: /tf/active/vicechatdev/document_controller_backup.py

• function complex_query_example 50.5% similar

  Executes a Neo4j Cypher query to find relationships between LIMS_Parameters nodes where either the source or target node name contains a specified search term.

  From: /tf/active/vicechatdev/neo4j_schema/neo4j_python_snippets.py