🔍 Code Extractor

Retrieves nodes from a Neo4j graph database that are related to a source node identified by its UID, with optional filtering by relationship type, target node label, and relationship direction.

Purpose

This function queries a Neo4j graph database to find all nodes connected to a specified source node. It supports flexible filtering by relationship type (e.g., 'KNOWS', 'WORKS_WITH'), target node labels (e.g., 'Person', 'Company'), and relationship direction (outgoing, incoming, or both). It's useful for traversing graph relationships, discovering connected entities, and building relationship maps in knowledge graphs or network analysis applications.

Source Code

def get_related_nodes(uid: str, 
                     relationship_type: Optional[str] = None,
                     target_label: Optional[str] = None,
                     direction: str = 'OUTGOING') -> List[Dict[str, Any]]:
    """
    Get nodes related to a node identified by UID.
    
    Args:
        uid: UID of source node
        relationship_type: Optional type of relationship to filter by
        target_label: Optional label of target nodes to filter by
        direction: 'OUTGOING', 'INCOMING', or 'BOTH'
        
    Returns:
        List of related nodes as dictionaries
    """
    try:
        # Determine relationship arrow direction
        if direction == 'OUTGOING':
            arrow = '-[r]->'
        elif direction == 'INCOMING':
            arrow = '<-[r]-'
        else:  # BOTH
            arrow = '-[r]-'
            
        # Add relationship type if specified
        if relationship_type:
            arrow = arrow.replace('r', f'r:{relationship_type}')
            
        # Add target label if specified
        target_pattern = '(t)'
        if target_label:
            target_pattern = f'(t:{target_label})'
            
        query = f"""
        MATCH (n {{UID: $uid}}) {arrow} {target_pattern}
        RETURN t
        """
        
        driver = get_driver()
        with driver.session() as session:
            result = session.run(query, uid=uid)
            
            nodes = []
            for record in result:
                node = record["t"]
                nodes.append(dict(node.items()))
                
            return nodes
            
    except Exception as e:
        logger.error(f"Error retrieving related nodes: {e}")
        return []

Parameters

Parameter Details

uid: Unique identifier (UID) of the source node from which to find related nodes. Must be a string that matches the UID property of an existing node in the Neo4j database.

relationship_type: Optional filter to restrict results to specific relationship types (e.g., 'FRIEND_OF', 'MANAGES'). If None, all relationship types are included. Must match Neo4j relationship type naming conventions (uppercase with underscores).

target_label: Optional filter to restrict results to nodes with a specific label (e.g., 'Person', 'Organization'). If None, nodes with any label are returned. Must match Neo4j node label naming conventions.

direction: Specifies the direction of relationships to traverse. Accepts 'OUTGOING' (default, follows relationships from source to target), 'INCOMING' (follows relationships pointing to the source), or 'BOTH' (follows relationships in either direction).

Return Value

Type: List[Dict[str, Any]]

Returns a List of dictionaries, where each dictionary represents a related node with its properties as key-value pairs. Returns an empty list if no related nodes are found or if an error occurs. Each dictionary contains all properties stored on the target node in Neo4j.

Dependencies

• neo4j
• logging
• typing

Required Imports

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

Usage Example

# Get all outgoing relationships from a node
related_nodes = get_related_nodes(uid='user-123')

# Get specific relationship type
friends = get_related_nodes(
    uid='user-123',
    relationship_type='FRIEND_OF',
    direction='OUTGOING'
)

# Get incoming relationships to Person nodes
managers = get_related_nodes(
    uid='employee-456',
    relationship_type='MANAGES',
    target_label='Person',
    direction='INCOMING'
)

# Get all connected nodes regardless of direction
all_connections = get_related_nodes(
    uid='node-789',
    direction='BOTH'
)

# Process results
for node in related_nodes:
    print(f"Node properties: {node}")
    if 'name' in node:
        print(f"Name: {node['name']}")

Best Practices

• Always handle the empty list return case, as it can indicate either no results or an error condition
• Check logs for error messages when an empty list is returned unexpectedly
• Ensure the source node with the specified UID exists before calling this function
• Use specific relationship_type and target_label filters when possible to improve query performance on large graphs
• Be cautious with direction='BOTH' on highly connected nodes as it may return large result sets
• The function uses parameterized queries to prevent Cypher injection attacks
• Consider pagination or limiting results for nodes with many relationships
• Ensure proper Neo4j driver session management is handled by the get_driver() function
• Relationship type and node label names should follow Neo4j naming conventions (uppercase for relationships, CamelCase for labels)

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function create_relationship 65.8% similar

  Creates a directed relationship between two Neo4j graph database nodes identified by their UIDs, with optional properties attached to the relationship.

  From: /tf/active/vicechatdev/CDocs/db/db_operations.py

• function get_node_by_uid 63.3% similar

  Retrieves a node from a Neo4j graph database by its unique identifier (UID) and returns it as a dictionary.

  From: /tf/active/vicechatdev/CDocs/db/db_operations.py

• function create_relationship_with_retry 58.3% similar

  Creates a relationship between two nodes in a graph database with automatic retry logic that handles case-insensitive UID matching when the initial attempt fails.

  From: /tf/active/vicechatdev/CDocs/utils/uid_helper.py

• function get_document_with_relationships 57.1% similar

  Retrieves a complete document from Neo4j graph database along with all its related entities including versions, reviews, approvals, and authors.

  From: /tf/active/vicechatdev/CDocs/db/db_operations.py

• function delete_node 55.9% similar

  Deletes a node from a Neo4j graph database by its unique identifier (UID), with optional cascade deletion of connected nodes and relationships.

  From: /tf/active/vicechatdev/CDocs/db/db_operations.py