🔍 Code Extractor

Retrieves notifications for a specific user from a Neo4j graph database, with options for filtering unread notifications and pagination support.

Purpose

This function queries a Neo4j database to fetch notifications associated with a user identified by their UID. It supports filtering for unread notifications only, pagination through skip/limit parameters, and automatically parses JSON-formatted notification details. The function is designed for notification management systems where users need to view their notifications with flexible filtering and pagination options.

Source Code

def get_user_notifications(user_uid: str,
                         unread_only: bool = False,
                         limit: int = 50,
                         skip: int = 0) -> List[Dict[str, Any]]:
    """
    Get notifications for a user.
    
    Args:
        user_uid: UID of user
        unread_only: Whether to return only unread notifications
        limit: Maximum number of notifications to return
        skip: Number of notifications to skip (for pagination)
        
    Returns:
        List of notification dictionaries
    """
    try:
        # Build query
        query = """
        MATCH (u:User {UID: $user_uid})-[:HAS_NOTIFICATION]->(n:Notification)
        """
        
        if unread_only:
            query += "WHERE n.read = false "
            
        query += f"""
        RETURN n
        ORDER BY n.timestamp DESC
        SKIP {skip} LIMIT {limit}
        """
        
        # Execute query
        result = db.run_query(query, {"user_uid": user_uid})
        
        # Convert to list of dictionaries
        notifications = []
        for record in result:
            notification = record['n']
            
            # Convert details from JSON string to dictionary if it exists
            if 'details' in notification and isinstance(notification['details'], str):
                try:
                    notification['details'] = json.loads(notification['details'])
                except:
                    pass
                    
            notifications.append(notification)
            
        return notifications
        
    except Exception as e:
        logger.error(f"Error getting notifications for user {user_uid}: {e}")
        return []

Parameters

Parameter Details

user_uid: String identifier (UID) of the user whose notifications should be retrieved. This is used to match against the User node's UID property in the Neo4j database.

unread_only: Boolean flag that when set to True, filters the results to return only notifications where the 'read' property is false. Defaults to False, which returns all notifications regardless of read status.

limit: Integer specifying the maximum number of notifications to return in a single query. Used for pagination. Defaults to 50. Should be a positive integer.

skip: Integer specifying how many notifications to skip before returning results. Used for pagination to retrieve subsequent pages of notifications. Defaults to 0 (start from the beginning).

Return Value

Type: List[Dict[str, Any]]

Returns a List of dictionaries, where each dictionary represents a notification with its properties. Each notification dictionary contains fields from the Neo4j Notification node, including a 'details' field that is automatically parsed from JSON string to dictionary if present. Returns an empty list if an error occurs or if no notifications are found. Notifications are ordered by timestamp in descending order (newest first).

Dependencies

• json
• logging
• typing
• CDocs.db
• CDocs.config.settings

Required Imports

import json
import logging
from typing import List, Dict, Any
from CDocs import db

Usage Example

# Get all notifications for a user
notifications = get_user_notifications(user_uid='user123')

# Get only unread notifications
unread_notifications = get_user_notifications(
    user_uid='user123',
    unread_only=True
)

# Get second page of notifications (pagination)
page_2_notifications = get_user_notifications(
    user_uid='user123',
    limit=20,
    skip=20
)

# Process notifications
for notification in notifications:
    print(f"Notification: {notification.get('message')}")
    if 'details' in notification:
        # details is automatically parsed from JSON
        print(f"Details: {notification['details']}")

Best Practices

• Always handle the empty list return case, as errors are caught and logged but return an empty list rather than raising exceptions
• Use pagination (limit and skip parameters) when dealing with users who may have many notifications to avoid memory issues
• The function uses string interpolation for skip and limit in the Cypher query, which is safe here since they are integers, but be aware of this pattern
• The details field JSON parsing failure is silently ignored (bare except clause), so verify the details field structure if it's critical to your application
• Ensure the logger is properly configured before calling this function to capture error messages
• The function assumes a specific Neo4j schema with User and Notification nodes connected by HAS_NOTIFICATION relationships
• Consider the performance implications of ORDER BY on large notification sets; ensure timestamp is indexed in Neo4j

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function count_user_notifications 86.7% similar

  Counts the total number of notifications for a specific user in a Neo4j graph database, with an option to filter for only unread notifications.

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

• function delete_all_notifications 72.0% similar

  Deletes all notifications or only read notifications for a specific user from a Neo4j graph database.

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

• function delete_notification 63.3% similar

  Deletes a notification node from a Neo4j graph database by its unique identifier (UID).

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

• function get_users 62.0% similar

  Retrieves user information from a Neo4j graph database for display in an admin panel, with optional filtering by user ID.

  From: /tf/active/vicechatdev/CDocs/controllers/admin_controller.py

• function get_node_by_uid 60.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