🔍 Code Extractor

Retrieves a comprehensive training overview for the admin panel, including training plans, active assignments, and recent completions from a Neo4j graph database.

Purpose

This function provides administrators with a complete view of the document training system. It queries the database for documents requiring training, calculates completion rates, retrieves active training assignments, and fetches recent completions. The data is aggregated and formatted for display in an administrative dashboard, enabling tracking of training compliance and progress across the organization.

Source Code

def get_training_overview(user: DocUser = None) -> Dict[str, Any]:
    """
    Get training overview for admin panel.
    
    Args:
        user: Admin user requesting overview
        
    Returns:
        Dictionary with training overview data
    """
    try:
        # Note: Permission checking is handled at the Flask route level for admin users
        logger.info(f"Getting training overview - user provided: {user is not None}")
        
        # Get all documents with training enabled and their assignments
        query_plans = """
        MATCH (d:ControlledDocument)
        WHERE d.training_required = true
        OPTIONAL MATCH (u:User)-[r:REQUIRES_TRAINING]->(d)
        WITH d, COUNT(r) as assigned_count, 
             COUNT(CASE WHEN r.status = 'COMPLETED' THEN 1 END) as completed_count
        RETURN d, assigned_count, completed_count
        ORDER BY d.docNumber
        """
        
        plans_result = db.run_query(query_plans)
        plans = []
        
        for record in plans_result:
            doc_data = record['d']
            assigned_count = record['assigned_count']
            completed_count = record['completed_count']
            
            completion_rate = (completed_count / assigned_count * 100) if assigned_count > 0 else 0
            
            plans.append({
                'uid': doc_data.get('UID'),
                'name': f"Training for {doc_data.get('title', 'Unknown Document')}",
                'document_uid': doc_data.get('UID'),
                'document_title': doc_data.get('title', 'Unknown Document'),
                'document_number': doc_data.get('docNumber', 'Unknown'),
                'document_count': 1,  # Each plan is for one document
                'validity_days': doc_data.get('training_validity_days', 365),
                'quiz_required': doc_data.get('training_quiz_required', False),
                'assigned_count': assigned_count,
                'completed_count': completed_count,
                'completion_rate': round(completion_rate, 1),
                'created': doc_data.get('training_enabled_date', 'Unknown'),
                'status': 'active' if assigned_count > 0 else 'inactive'
            })
        
        # Get active assignments
        query_assignments = """
        MATCH (u:User)-[r:REQUIRES_TRAINING]->(d:ControlledDocument)
        WHERE r.status IN ['REQUIRED', 'IN_PROGRESS']
        RETURN u, r, d
        ORDER BY r.assigned_date DESC
        LIMIT 50
        """
        
        logger.info("Executing assignments query...")
        assignments_result = db.run_query(query_assignments)
        logger.info(f"Assignments query returned {len(list(assignments_result)) if assignments_result else 0} results")
        assignments = []
        
        # Re-run the query since we consumed the iterator above
        assignments_result = db.run_query(query_assignments)
        
        for record in assignments_result:
            user_data = record['u']
            rel_data = record['r']
            doc_data = record['d']
            
            assignments.append({
                'uid': f"{user_data.get('UID', '')}-{doc_data.get('UID', '')}",
                'user_name': user_data.get('Name', user_data.get('username', 'Unknown')),
                'user_email': user_data.get('Mail', ''),
                'document_title': doc_data.get('title', 'Unknown Document'),
                'document_number': doc_data.get('docNumber', 'Unknown'),
                'assigned_date': rel_data.get('assigned_date'),
                'expires_date': rel_data.get('expires_date'),
                'status': rel_data.get('status', 'REQUIRED'),
                'assigned_by': rel_data.get('assigned_by', 'Unknown')
            })
            
        logger.info(f"Processed {len(assignments)} assignments")
        
        # Get recent completions
        query_completions = """
        MATCH (u:User)-[r:REQUIRES_TRAINING]->(d:ControlledDocument)
        WHERE r.status = 'COMPLETED'
        RETURN u, r, d
        ORDER BY r.completed_date DESC
        LIMIT 25
        """
        
        logger.info("Executing completions query...")
        completions_result = db.run_query(query_completions)
        completions = []
        
        for record in completions_result:
            user_data = record['u']
            rel_data = record['r']
            doc_data = record['d']
            
            completions.append({
                'uid': f"{user_data.get('UID', '')}-{doc_data.get('UID', '')}",
                'user_name': user_data.get('Name', user_data.get('username', 'Unknown')),
                'user_email': user_data.get('Mail', ''),
                'document_title': doc_data.get('title', 'Unknown Document'),
                'document_number': doc_data.get('docNumber', 'Unknown'),
                'completed_date': rel_data.get('completed_date'),
                'score': rel_data.get('score', 'N/A'),
                'assigned_date': rel_data.get('assigned_date')
            })
            
        logger.info(f"Processed {len(completions)} completions")
        
        logger.info(f"Final training overview: {len(plans)} plans, {len(assignments)} assignments, {len(completions)} completions")
        
        return {
            "success": True,
            "plans": plans,
            "assignments": assignments,
            "completions": completions,
            "summary": {
                "total_plans": len(plans),
                "active_assignments": len(assignments),
                "recent_completions": len(completions)
            }
        }
        
    except Exception as e:
        logger.error(f"Error getting training overview: {e}")
        return {
            "success": False,
            "message": f"Failed to get training overview: {e}",
            "plans": [],
            "assignments": [],
            "completions": [],
            "summary": {
                "total_plans": 0,
                "active_assignments": 0,
                "recent_completions": 0
            }
        }

Parameters

Parameter Details

user: Optional DocUser object representing the admin user requesting the overview. Defaults to None. Note that permission checking is handled at the Flask route level, so this parameter is primarily used for logging purposes to track which admin accessed the overview.

Return Value

Type: Dict[str, Any]

Returns a Dict[str, Any] containing: 'success' (bool indicating operation success), 'plans' (list of training plan dictionaries with document info, assignment counts, completion rates), 'assignments' (list of up to 50 active training assignments with user and document details), 'completions' (list of up to 25 recent completed trainings), 'summary' (dict with aggregate counts), and 'message' (error message if success is False). Each plan includes UID, document details, validity days, quiz requirements, and completion statistics. Assignments and completions include user information, document details, dates, and status.

Dependencies

• typing
• datetime
• logging
• traceback
• CDocs.config
• CDocs.models.document
• CDocs.models.user_extensions
• CDocs.models.training
• CDocs.utils
• CDocs.db
• CDocs.controllers

Required Imports

from typing import Dict, Any
import logging
from CDocs.models.user_extensions import DocUser
from CDocs.db import db_operations as db
from CDocs.controllers import log_controller_action

Usage Example

from CDocs.models.user_extensions import DocUser
from CDocs.controllers.training_controller import get_training_overview

# Get training overview (typically called from admin route)
admin_user = DocUser(uid='admin123', username='admin')
result = get_training_overview(user=admin_user)

if result['success']:
    print(f"Total training plans: {result['summary']['total_plans']}")
    print(f"Active assignments: {result['summary']['active_assignments']}")
    print(f"Recent completions: {result['summary']['recent_completions']}")
    
    # Access training plans
    for plan in result['plans']:
        print(f"Document: {plan['document_title']} - Completion: {plan['completion_rate']}%")
    
    # Access active assignments
    for assignment in result['assignments']:
        print(f"{assignment['user_name']} assigned to {assignment['document_title']}")
    
    # Access recent completions
    for completion in result['completions']:
        print(f"{completion['user_name']} completed {completion['document_title']} with score {completion['score']}")
else:
    print(f"Error: {result['message']}")

Best Practices

• This function is decorated with @log_controller_action('get_training_overview') for audit logging
• Permission checking must be implemented at the Flask route level before calling this function
• The function handles database query failures gracefully and returns a structured error response
• Query results are limited (50 assignments, 25 completions) to prevent performance issues with large datasets
• The function consumes the assignments_result iterator twice, requiring re-execution of the query
• All database exceptions are caught and logged with detailed error messages
• Completion rates are calculated with division-by-zero protection
• The user parameter is optional and primarily used for logging; actual authorization should occur before calling
• Consider implementing pagination for assignments and completions if the limits become restrictive
• The function assumes specific Neo4j node labels (ControlledDocument, User) and relationship types (REQUIRES_TRAINING)
• Status values are expected to be 'REQUIRED', 'IN_PROGRESS', or 'COMPLETED'

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function get_user_training_dashboard 80.4% similar

  Retrieves a user's training dashboard data by querying Neo4j for required and completed training records associated with controlled documents.

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

• function get_document_training_info 69.4% similar

  Retrieves comprehensive training information for a specific controlled document, including training configuration, assigned users, completion status, and statistics.

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

• function get_training_assignment 67.0% similar

  Retrieves a specific training assignment for a user from a Neo4j graph database, validating user authorization and parsing a composite UID format.

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

• function complete_user_training_by_uid 63.8% similar

  Completes a user's training assignment for a controlled document by updating the training relationship status, recording completion date and score, and logging the event to the audit trail.

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

• class UserTraining 63.1% similar

  A model class representing a user's training status for a specific controlled document, managing training assignments, completion tracking, and expiration dates.

  From: /tf/active/vicechatdev/CDocs/models/training.py