🔍 Code Extractor

Creates an in-app notification for a user in a graph database, linking it to the user and optionally to a related resource.

Purpose

This function creates notifications within an application's notification system. It validates the notification type against a predefined set of types, generates a unique notification UID, stores notification metadata in a Neo4j graph database, and establishes relationships between the user, notification, and any associated resource. It's designed for event-driven notification systems where users need to be alerted about various application events.

Source Code

def create_notification(notification_type: str,
                       user_uid: str,
                       resource_uid: Optional[str] = None,
                       resource_type: Optional[str] = None,
                       message: Optional[str] = None,
                       details: Optional[Dict[str, Any]] = None) -> Optional[str]:
    """
    Create an in-app notification.
    
    Args:
        notification_type: Type of notification from NOTIFICATION_TYPES
        user_uid: UID of user to notify
        resource_uid: Optional UID of resource this notification is about
        resource_type: Optional type of resource
        message: Optional message text
        details: Optional additional details
        
    Returns:
        UID of created notification or None if creation failed
    """
    try:
        # Validate notification type
        if notification_type not in NOTIFICATION_TYPES:
            logger.error(f"Invalid notification type: {notification_type}")
            return None
            
        # Get notification info
        notification_info = NOTIFICATION_TYPES[notification_type]
        
        # Prepare notification properties
        notification_uid = str(uuid.uuid4())
        props = {
            'UID': notification_uid,
            'type': notification_type,
            'category': notification_info['category'],
            'priority': notification_info['priority'],
            'icon': notification_info.get('icon', 'bell'),
            'timestamp': datetime.now(),
            'read': False
        }
        
        if message:
            props['message'] = message
            
        if resource_uid:
            props['resourceUID'] = resource_uid
            
        if resource_type:
            props['resourceType'] = resource_type
            
        if details:
            props['details'] = json.dumps(details)
            
        # Create notification node
        db.run_query(
            """
            MATCH (u:User {UID: $user_uid})
            CREATE (u)-[:HAS_NOTIFICATION]->(n:Notification)
            SET n = $props
            RETURN n.UID as notificationUID
            """,
            {"user_uid": user_uid, "props": props}
        )
        
        # Link to resource if resource UID provided
        if resource_uid:
            db.run_query(
                """
                MATCH (n:Notification {UID: $notification_uid})
                MATCH (r {UID: $resource_uid})
                CREATE (n)-[:REFERS_TO]->(r)
                """,
                {"notification_uid": notification_uid, "resource_uid": resource_uid}
            )
            
        return notification_uid
        
    except Exception as e:
        logger.error(f"Error creating notification: {e}")
        return None

Parameters

Parameter Details

notification_type: String identifier for the type of notification. Must be a valid key in the NOTIFICATION_TYPES dictionary (not shown in code but referenced). This determines the category, priority, and icon for the notification.

user_uid: Unique identifier (UID) string for the user who should receive this notification. Must correspond to an existing User node in the database.

resource_uid: Optional unique identifier string for a resource that this notification relates to (e.g., a document, review, or approval). If provided, a REFERS_TO relationship will be created between the notification and the resource node.

resource_type: Optional string describing the type of resource being referenced (e.g., 'document', 'review', 'approval'). Used for categorization and filtering.

message: Optional custom message text to include in the notification. If not provided, the notification may use a default message based on the notification type.

details: Optional dictionary containing additional structured data about the notification. Will be JSON-serialized and stored in the notification node.

Return Value

Type: Optional[str]

Returns a string containing the UUID of the newly created notification if successful, or None if the creation failed due to validation errors, database errors, or exceptions. The UUID can be used to reference, update, or delete the notification later.

Dependencies

• uuid
• json
• datetime
• logging
• typing

Required Imports

import uuid
import json
from datetime import datetime
from typing import Dict, Any, Optional
import logging

Usage Example

# Assuming NOTIFICATION_TYPES, logger, and db are configured

# Create a simple notification
notification_uid = create_notification(
    notification_type='document_shared',
    user_uid='user-123-456',
    message='A document has been shared with you'
)

if notification_uid:
    print(f'Notification created: {notification_uid}')
else:
    print('Failed to create notification')

# Create a notification with resource link and details
notification_uid = create_notification(
    notification_type='review_assigned',
    user_uid='user-789-012',
    resource_uid='doc-abc-def',
    resource_type='document',
    message='You have been assigned to review a document',
    details={
        'document_title': 'Q4 Report',
        'due_date': '2024-01-15',
        'assigned_by': 'manager-uid-123'
    }
)

if notification_uid:
    print(f'Review notification created: {notification_uid}')

Best Practices

• Ensure NOTIFICATION_TYPES dictionary is properly defined before calling this function
• Always check the return value for None to handle creation failures gracefully
• Validate user_uid exists in the database before calling to avoid silent failures
• If providing resource_uid, ensure the resource node exists in the database
• Keep the details dictionary serializable (avoid complex objects that can't be JSON-encoded)
• Use appropriate notification_type values that exist in NOTIFICATION_TYPES to avoid validation errors
• Consider implementing retry logic for transient database connection failures
• Monitor logs for error messages to track notification creation failures
• The function creates two separate database queries when resource_uid is provided; consider transaction handling for atomicity

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function send_notification 74.8% similar

  Sends in-app notifications to one or more users and optionally sends corresponding email notifications using templates.

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

• function delete_notification 64.6% similar

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

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

• function log_event 58.2% similar

  Creates and persists an audit trail event in a Neo4j graph database, linking it to users, resources, and an audit trail node with comprehensive event metadata.

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

• function get_user_notifications 57.8% similar

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

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

• function notify_document_update 55.5% similar

  Sends notifications to interested users (owners, department members, subscribers, and users with permissions) when a controlled document is created, updated, or has a status/version change.

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