🔍 Code Extractor

Central manager class for handling in-app and email notifications within the application, providing methods for creating, managing, and sending notifications to users.

Purpose

NotificationManager serves as a facade/wrapper class that centralizes all notification-related operations in the application. It manages both in-app notifications (stored in database) and email notifications. The class provides methods for creating notifications, marking them as read/unread, retrieving user notifications with pagination, counting notifications, deleting notifications, and sending notifications to users with optional email delivery. It also includes specialized methods for notifying users about reviews, approvals, and document updates, as well as scheduling reminder notifications. The class acts as a high-level interface that delegates to lower-level notification functions while maintaining logging and state.

Source Code

class NotificationManager:
    """
    Central manager for handling notifications within the application.
    
    This class provides methods for creating, managing, and sending
    notifications to users through both in-app and email channels.
    """
    
    def __init__(self):
        """Initialize the notification manager."""
        self.email_templates = get_email_templates()
        self.logger = logging.getLogger('CDocs.utils.notifications.NotificationManager')
        self.logger.info("NotificationManager initialized")
    
    def create_notification(self, 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
        """
        return create_notification(notification_type, user_uid, resource_uid, resource_type, message, details)
    
    def mark_notification_read(self, notification_uid: str, read: bool = True) -> bool:
        """
        Mark a notification as read or unread.
        
        Args:
            notification_uid: UID of notification
            read: Whether to mark as read (True) or unread (False)
            
        Returns:
            Boolean indicating success
        """
        return mark_notification_read(notification_uid, read)
    
    def get_user_notifications(self, 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
        """
        return get_user_notifications(user_uid, unread_only, limit, skip)
    
    def count_user_notifications(self, user_uid: str, unread_only: bool = False) -> int:
        """
        Count notifications for a user.
        
        Args:
            user_uid: UID of user
            unread_only: Whether to count only unread notifications
            
        Returns:
            Number of notifications
        """
        return count_user_notifications(user_uid, unread_only)
    
    def delete_notification(self, notification_uid: str) -> bool:
        """
        Delete a notification.
        
        Args:
            notification_uid: UID of notification
            
        Returns:
            Boolean indicating success
        """
        return delete_notification(notification_uid)
    
    def delete_all_notifications(self, user_uid: str, read_only: bool = False) -> bool:
        """
        Delete all notifications for a user.
        
        Args:
            user_uid: UID of user
            read_only: Whether to delete only read notifications
            
        Returns:
            Boolean indicating success
        """
        return delete_all_notifications(user_uid, read_only)
    
    def send_notification(self, notification_type: str,
                        users: Union[List[Union[DocUser, str]], DocUser, str],
                        resource_uid: Optional[str] = None,
                        resource_type: Optional[str] = None,
                        message: Optional[str] = None,
                        details: Optional[Dict[str, Any]] = None,
                        send_email: bool = False,
                        email_template: Optional[str] = None,
                        email_data: Optional[Dict[str, Any]] = None) -> List[str]:
        """
        Send a notification to one or more users, optionally with email.
        
        Args:
            notification_type: Type of notification from NOTIFICATION_TYPES
            users: User(s) to notify (DocUser instance(s) or UID(s))
            resource_uid: Optional UID of resource this notification is about
            resource_type: Optional type of resource
            message: Optional message text
            details: Optional additional details
            send_email: Whether to also send email notification
            email_template: Template to use for email (if sending email)
            email_data: Data for email template (if sending email)
            
        Returns:
            List of created notification UIDs
        """
        return send_notification(
            notification_type, users, resource_uid, resource_type, 
            message, details, send_email, email_template, email_data
        )
    
    def notify_review(self, review_cycle,
                    notification_type: str,
                    reviewer_uids: Optional[List[str]] = None) -> List[str]:
        """
        Send review notifications to reviewers.
        
        Args:
            review_cycle: ReviewCycle instance
            notification_type: Type of notification
            reviewer_uids: Optional specific reviewers to notify (default: all)
            
        Returns:
            List of created notification UIDs
        """
        return notify_review(review_cycle, notification_type, reviewer_uids)
    
    def notify_approval(self, approval,
                      notification_type: str,
                      approver_uids: Optional[List[str]] = None) -> List[str]:
        """
        Send approval notifications to approvers.
        
        Args:
            approval: Approval instance
            notification_type: Type of notification
            approver_uids: Optional specific approvers to notify (default: all)
            
        Returns:
            List of created notification UIDs
        """
        return notify_approval(approval, notification_type, approver_uids)
    
    def notify_document_update(self, document, notification_type: str) -> List[str]:
        """
        Send document update notifications to interested users.
        
        Args:
            document: ControlledDocument instance
            notification_type: Type of notification
            
        Returns:
            List of created notification UIDs
        """
        return notify_document_update(document, notification_type)
    
    def schedule_reminders(self):
        """
        Schedule reminder notifications for upcoming reviews and approvals.
        This should be called by a scheduled task daily.
        """
        schedule_reminders()

Parameters

Name	Type	Default	Kind
bases	-	-

Parameter Details

__init__: The constructor takes no parameters. It initializes the notification manager by loading email templates via get_email_templates() and setting up a logger instance for the NotificationManager component.

Return Value

Instantiation returns a NotificationManager instance. Methods return various types: create_notification and send_notification return notification UIDs (Optional[str] or List[str]), mark_notification_read, delete_notification, and delete_all_notifications return boolean success indicators, get_user_notifications returns a list of notification dictionaries, count_user_notifications returns an integer count, and schedule_reminders returns None.

Class Interface

Methods

Attributes

Name	Type	Description	Scope
email_templates	Dict or similar structure	Stores email templates loaded during initialization via get_email_templates() function	instance
logger	logging.Logger	Logger instance for the NotificationManager component, used to log initialization and operations	instance

Dependencies

• logging
• typing
• CDocs
• CDocs.config
• CDocs.models.user_extensions
• CDocs.utils
• CDocs.controllers.document_controller
• models.review
• models.approval

Required Imports

import logging
from typing import Dict, List, Any, Optional, Union
from CDocs.models.user_extensions import DocUser
from models.review import ReviewCycle
from models.approval import Approval

Usage Example

# Instantiate the notification manager
notif_manager = NotificationManager()

# Create a simple in-app notification
notif_uid = notif_manager.create_notification(
    notification_type='DOCUMENT_UPDATED',
    user_uid='user123',
    resource_uid='doc456',
    resource_type='document',
    message='Your document has been updated'
)

# Send notification to multiple users with email
notif_uids = notif_manager.send_notification(
    notification_type='REVIEW_REQUESTED',
    users=['user123', 'user456'],
    resource_uid='doc789',
    message='Please review this document',
    send_email=True,
    email_template='review_request',
    email_data={'document_title': 'Safety Manual'}
)

# Get unread notifications for a user
unread_notifs = notif_manager.get_user_notifications(
    user_uid='user123',
    unread_only=True,
    limit=20
)

# Mark notification as read
notif_manager.mark_notification_read(notif_uid, read=True)

# Count unread notifications
count = notif_manager.count_user_notifications('user123', unread_only=True)

# Delete a notification
notif_manager.delete_notification(notif_uid)

# Notify reviewers about a review cycle
review_notif_uids = notif_manager.notify_review(
    review_cycle=review_cycle_instance,
    notification_type='REVIEW_ASSIGNED'
)

# Schedule reminders (typically called by scheduled task)
notif_manager.schedule_reminders()

Best Practices

• Instantiate NotificationManager once and reuse the instance throughout the application lifecycle to avoid repeated initialization of email templates and logger
• Always check return values from create_notification and send_notification to verify notifications were created successfully (None indicates failure)
• Use unread_only=True when querying notifications to reduce database load and improve performance
• Implement pagination using limit and skip parameters when retrieving notifications to avoid loading large datasets
• When sending notifications with email, ensure email_template and email_data are properly provided to avoid email sending failures
• The schedule_reminders method should be called by a scheduled task (e.g., daily cron job) rather than in request handlers
• Use the specialized methods (notify_review, notify_approval, notify_document_update) for domain-specific notifications as they handle proper context and recipients
• The class is stateless except for email_templates and logger, making it safe for concurrent use
• All methods delegate to module-level functions, so this class primarily provides a convenient interface and logging
• When deleting notifications, use read_only=True to preserve unread notifications if needed
• The users parameter in send_notification accepts both DocUser instances and UID strings for flexibility

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function send_notification 53.6% 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 notify_approval 50.5% similar

  Sends approval notifications to designated approvers for a document version, supporting multiple notification types (requested, reminder, overdue, completed, rejected, signed) with email and in-app notifications.

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

• class BaseUIComponent 49.8% similar

  Base class for UI components that provides common notification functionality for displaying error, success, warning, and info messages.

  From: /tf/active/vicechatdev/CDocs/ui/base_ui.py

• class CDocsApp 47.4% similar

  Panel-based web application class for the CDocs Controlled Document System that provides a complete UI with navigation, authentication, and document management features.

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

• class SignatureManager 46.8% similar

  A class that manages digital signature images for documents, providing functionality to store, retrieve, and list signature files in a designated directory.

  From: /tf/active/vicechatdev/document_auditor/src/security/signature_manager.py

class NotificationManager:
    """
    Central manager for handling notifications within the application.
    
    This class provides methods for creating, managing, and sending
    notifications to users through both in-app and email channels.
    """
    
    def __init__(self):
        """Initialize the notification manager."""
        self.email_templates = get_email_templates()
        self.logger = logging.getLogger('CDocs.utils.notifications.NotificationManager')
        self.logger.info("NotificationManager initialized")
    
    def create_notification(self, 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
        """
        return create_notification(notification_type, user_uid, resource_uid, resource_type, message, details)
    
    def mark_notification_read(self, notification_uid: str, read: bool = True) -> bool:
        """
        Mark a notification as read or unread.
        
        Args:
            notification_uid: UID of notification
            read: Whether to mark as read (True) or unread (False)
            
        Returns:
            Boolean indicating success
        """
        return mark_notification_read(notification_uid, read)
    
    def get_user_notifications(self, 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
        """
        return get_user_notifications(user_uid, unread_only, limit, skip)
    
    def count_user_notifications(self, user_uid: str, unread_only: bool = False) -> int:
        """
        Count notifications for a user.
        
        Args:
            user_uid: UID of user
            unread_only: Whether to count only unread notifications
            
        Returns:
            Number of notifications
        """
        return count_user_notifications(user_uid, unread_only)
    
    def delete_notification(self, notification_uid: str) -> bool:
        """
        Delete a notification.
        
        Args:
            notification_uid: UID of notification
            
        Returns:
            Boolean indicating success
        """
        return delete_notification(notification_uid)
    
    def delete_all_notifications(self, user_uid: str, read_only: bool = False) -> bool:
        """
        Delete all notifications for a user.
        
        Args:
            user_uid: UID of user
            read_only: Whether to delete only read notifications
            
        Returns:
            Boolean indicating success
        """
        return delete_all_notifications(user_uid, read_only)
    
    def send_notification(self, notification_type: str,
                        users: Union[List[Union[DocUser, str]], DocUser, str],
                        resource_uid: Optional[str] = None,
                        resource_type: Optional[str] = None,
                        message: Optional[str] = None,
                        details: Optional[Dict[str, Any]] = None,
                        send_email: bool = False,
                        email_template: Optional[str] = None,
                        email_data: Optional[Dict[str, Any]] = None) -> List[str]:
        """
        Send a notification to one or more users, optionally with email.
        
        Args:
            notification_type: Type of notification from NOTIFICATION_TYPES
            users: User(s) to notify (DocUser instance(s) or UID(s))
            resource_uid: Optional UID of resource this notification is about
            resource_type: Optional type of resource
            message: Optional message text
            details: Optional additional details
            send_email: Whether to also send email notification
            email_template: Template to use for email (if sending email)
            email_data: Data for email template (if sending email)
            
        Returns:
            List of created notification UIDs
        """
        return send_notification(
            notification_type, users, resource_uid, resource_type, 
            message, details, send_email, email_template, email_data
        )
    
    def notify_review(self, review_cycle,
                    notification_type: str,
                    reviewer_uids: Optional[List[str]] = None) -> List[str]:
        """
        Send review notifications to reviewers.
        
        Args:
            review_cycle: ReviewCycle instance
            notification_type: Type of notification
            reviewer_uids: Optional specific reviewers to notify (default: all)
            
        Returns:
            List of created notification UIDs
        """
        return notify_review(review_cycle, notification_type, reviewer_uids)
    
    def notify_approval(self, approval,
                      notification_type: str,
                      approver_uids: Optional[List[str]] = None) -> List[str]:
        """
        Send approval notifications to approvers.
        
        Args:
            approval: Approval instance
            notification_type: Type of notification
            approver_uids: Optional specific approvers to notify (default: all)
            
        Returns:
            List of created notification UIDs
        """
        return notify_approval(approval, notification_type, approver_uids)
    
    def notify_document_update(self, document, notification_type: str) -> List[str]:
        """
        Send document update notifications to interested users.
        
        Args:
            document: ControlledDocument instance
            notification_type: Type of notification
            
        Returns:
            List of created notification UIDs
        """
        return notify_document_update(document, notification_type)
    
    def schedule_reminders(self):
        """
        Schedule reminder notifications for upcoming reviews and approvals.
        This should be called by a scheduled task daily.
        """
        schedule_reminders()