🔍 Code Extractor

Retrieves nodes from a Neo4j graph database by label with optional property filtering, pagination, and sorting capabilities.

Purpose

This function queries a Neo4j database to fetch nodes matching a specific label. It supports filtering by node properties, pagination through skip/limit parameters, and ordering results by a specified property. It's designed for flexible node retrieval in graph database applications, handling common query patterns like filtered searches and paginated results. The function returns nodes as dictionaries for easy manipulation and includes error handling that returns an empty list on failure.

Source Code

def get_nodes_by_label(label: str, 
                      properties: Optional[Dict[str, Any]] = None,
                      limit: int = 100,
                      skip: int = 0,
                      order_by: Optional[str] = None,
                      descending: bool = False) -> List[Dict[str, Any]]:
    """
    Get nodes by label with optional filtering by properties.
    
    Args:
        label: Node label to match
        properties: Optional property filters
        limit: Maximum number of nodes to return
        skip: Number of nodes to skip (for pagination)
        order_by: Optional property to order results by
        descending: Whether to order in descending order
        
    Returns:
        List of matching nodes as dictionaries
    """
    try:
        if properties is None:
            properties = {}
            
        # Build WHERE clause from properties
        where_clauses = []
        for key, value in properties.items():
            where_clauses.append(f"n.{key} = ${key}")
            
        where_clause = " AND ".join(where_clauses) if where_clauses else "TRUE"
        
        # Build ORDER BY clause
        order_clause = ""
        if order_by:
            direction = "DESC" if descending else "ASC"
            order_clause = f"ORDER BY n.{order_by} {direction}"
            
        query = f"""
        MATCH (n:{label})
        WHERE {where_clause}
        {order_clause}
        RETURN n
        SKIP {skip} LIMIT {limit}
        """
        
        driver = get_driver()
        with driver.session() as session:
            result = session.run(query, **properties)
            
            nodes = []
            for record in result:
                node = record["n"]
                nodes.append(dict(node.items()))
                
            return nodes
            
    except Exception as e:
        logger.error(f"Error retrieving nodes by label: {e}")
        return []

Parameters

Parameter Details

label: The Neo4j node label to match (e.g., 'User', 'Document'). This is used in the MATCH clause to identify which type of nodes to retrieve. Must be a valid Neo4j label string.

properties: Optional dictionary of property key-value pairs to filter nodes. Each key-value pair becomes a WHERE clause condition (e.g., {'name': 'John', 'age': 30} filters for nodes where n.name='John' AND n.age=30). Defaults to None (no filtering).

limit: Maximum number of nodes to return in the result set. Used for pagination and controlling result size. Defaults to 100. Must be a positive integer.

skip: Number of nodes to skip before returning results. Used for pagination (e.g., skip=100 with limit=100 gets the second page). Defaults to 0. Must be a non-negative integer.

order_by: Optional property name to sort results by (e.g., 'created_at', 'name'). If None, results are returned in database order. Must be a valid property name on the nodes.

descending: Boolean flag to control sort order. If True, sorts in descending order; if False, sorts in ascending order. Only applies when order_by is specified. Defaults to False.

Return Value

Type: List[Dict[str, Any]]

Returns a List[Dict[str, Any]] containing matching nodes. Each node is represented as a dictionary with property names as keys and property values as values. Returns an empty list if no nodes match the criteria or if an error occurs. The list respects the limit and skip parameters for pagination.

Dependencies

• neo4j
• logging
• typing
• datetime
• uuid
• traceback

Required Imports

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

Usage Example

# Basic usage - get all User nodes
users = get_nodes_by_label('User')

# Filter by properties
active_users = get_nodes_by_label(
    'User',
    properties={'status': 'active', 'verified': True}
)

# Pagination - get second page of 50 users
page_2_users = get_nodes_by_label(
    'User',
    limit=50,
    skip=50
)

# Sorted results
sorted_users = get_nodes_by_label(
    'User',
    order_by='created_at',
    descending=True,
    limit=10
)

# Complex query with all parameters
filtered_sorted = get_nodes_by_label(
    'Document',
    properties={'category': 'technical', 'published': True},
    limit=20,
    skip=0,
    order_by='updated_at',
    descending=True
)

# Process results
for node in filtered_sorted:
    print(f"Node ID: {node.get('id')}, Title: {node.get('title')}")

Best Practices

• Always handle the possibility of an empty list return value, which can indicate either no matches or an error
• Use pagination (limit and skip) when dealing with potentially large result sets to avoid memory issues
• Ensure the label parameter matches existing node labels in your Neo4j schema to avoid empty results
• Property filter values should match the data types stored in Neo4j (e.g., don't pass string '123' if the property is stored as integer 123)
• Be cautious with order_by parameter - ensure the property exists on all nodes with that label to avoid inconsistent results
• Consider implementing retry logic or checking logger output for error details when empty lists are returned unexpectedly
• The function builds Cypher queries dynamically, so ensure label and order_by values are from trusted sources to prevent injection attacks
• Use appropriate limit values to balance between performance and completeness of results
• When using property filters, remember that all conditions are combined with AND logic

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function check_node_exists 65.5% similar

  Checks if a node with a specified label and matching properties exists in a Neo4j graph database.

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

• function get_node_labels 58.9% similar

  Retrieves all non-private attributes from the NodeLabels class as a dictionary, filtering out attributes that start with underscore.

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

• function get_all_parameter_medicationtypeproperties 57.8% similar

  Retrieves Parameter_MedicationTypeProperties nodes from a Neo4j graph database with a configurable limit on the number of results returned.

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

• function get_all_lims_sampletypeproperties 57.1% similar

  Queries a Neo4j graph database to retrieve LIMS_SampleTypeProperties nodes with a configurable limit on the number of results returned.

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

• function batch_create_nodes 56.4% similar

  Creates multiple Neo4j graph database nodes in batches for improved performance, automatically generating UIDs and timestamps for each node.

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