🔍 Code Extractor

A model class that manages training requirements and assignments for controlled documents, including enabling/disabling training, assigning training to users, and tracking training status.

Purpose

DocumentTraining is responsible for managing all aspects of document training requirements in a controlled document system. It handles enabling/disabling training requirements for documents, assigning training to specific users, tracking training status, managing training validity periods, and resetting training when documents are republished. The class interacts with a Neo4j graph database to persist training data and relationships between documents and users.

Source Code

class DocumentTraining(BaseModel):
    """
    Model representing training requirements for a document.
    """
    
    def __init__(self, document_uid: str = None, data: Dict[str, Any] = None):
        """Initialize DocumentTraining instance."""
        self.document_uid = document_uid
        if data:
            super().__init__(data)
        elif document_uid:
            # Load training data for document
            training_data = self._load_training_data()
            super().__init__(training_data or {})
        else:
            super().__init__({})
    
    def _load_training_data(self) -> Optional[Dict[str, Any]]:
        """Load training configuration for the document."""
        if not self.document_uid:
            return None
            
        query = """
        MATCH (d:ControlledDocument {UID: $doc_uid})
        RETURN d.training_required as required,
               d.training_validity_days as validity_days,
               d.training_quiz_required as quiz_required,
               d.training_instructions as instructions
        """
        
        result = db.run_query(query, {"doc_uid": self.document_uid})
        if result:
            return {
                "document_uid": self.document_uid,
                "training_required": result[0].get("required", False),
                "validity_days": result[0].get("validity_days", 365),
                "quiz_required": result[0].get("quiz_required", False),
                "instructions": result[0].get("instructions", "")
            }
        return None
    
    def enable_training(self, validity_days: int = 365, quiz_required: bool = False, instructions: str = "") -> bool:
        """Enable training for this document instance"""
        try:
            query = """
            MATCH (d:ControlledDocument {UID: $doc_uid}) 
            SET d.training_required = true,
                d.training_validity_days = $validity_days,
                d.training_quiz_required = $quiz_required,
                d.training_instructions = $instructions,
                d.training_enabled_date = datetime()
            RETURN d
            """
            
            result = db.run_query(query, {
                'doc_uid': self.document_uid,
                'validity_days': validity_days,
                'quiz_required': quiz_required,
                'instructions': instructions
            })
            
            if result:
                # Update local data
                self._data["training_required"] = True
                self._data["validity_days"] = validity_days
                self._data["quiz_required"] = quiz_required
                self._data["instructions"] = instructions
                return True
            return False
            
        except Exception as e:
            logger.error(f"Error enabling training for document {self.document_uid}: {e}")
            return False
    
    @staticmethod
    def enable_training(document_uid):
        """Enable training for a document"""
        print(f"DocumentTraining.enable_training called with document_uid: {document_uid}")
        try:
            # Check if document exists
            check_query = "MATCH (d:ControlledDocument {UID: $doc_uid}) RETURN d.UID as uid"
            print(f"Executing check query: {check_query} with doc_uid: {document_uid}")
            result = db.run_query(check_query, doc_uid=document_uid)
            print(f"Document check result: {result}")
            
            if not result:
                print(f"Document with UID {document_uid} not found")
                return False
            
            # Set training_required property (changed from training_enabled)
            query = "MATCH (d:ControlledDocument {UID: $doc_uid}) SET d.training_required = true RETURN d"
            print(f"Executing enable query: {query} with doc_uid: {document_uid}")
            result = db.run_query(query, doc_uid=document_uid)
            print(f"Update result: {result}")
            
            if result:
                print(f"Successfully enabled training for document {document_uid}")
                return True
            else:
                print(f"Failed to update document {document_uid}")
                return False
                        
        except Exception as e:
            print(f"Error enabling training for document {document_uid}: {str(e)}")
            return False
    
    def disable_training(self) -> bool:
        """Disable training requirement for the document."""
        try:
            query = """
            MATCH (d:ControlledDocument {UID: $doc_uid})
            SET d.training_required = false,
                d.training_disabled_date = datetime()
            RETURN d
            """
            
            result = db.run_query(query, {"doc_uid": self.document_uid})
            
            if result:
                self._data["training_required"] = False
                return True
            return False
            
        except Exception as e:
            logger.error(f"Error disabling training for document {self.document_uid}: {e}")
            return False
    
    def assign_user_training(self, user_uid: str, assigned_by: str, validity_days: int = 365) -> bool:
        """Assign training to a specific user."""
        try:
            from datetime import datetime, timedelta
            from CDocs.db import db_operations
            
            logger.info(f"Assigning training for document {self.document_uid} to user {user_uid}")
            
            # Calculate expiry date
            assigned_date = datetime.now()
            expires_date = assigned_date + timedelta(days=validity_days)
            
            # Create or update training assignment - FIXED: Use ControlledDocument label
            query = """
            MERGE (d:ControlledDocument {UID: $document_uid})
            MERGE (u:User {UID: $user_uid})
            MERGE (assigner:User {UID: $assigned_by})
            MERGE (u)-[r:REQUIRES_TRAINING]->(d)
            SET r.status = 'REQUIRED',
                r.assigned_date = $assigned_date,
                r.assigned_by = $assigned_by,
                r.expires_date = $expires_date,
                r.validity_days = $validity_days,
                r.updated_date = $assigned_date
            RETURN r
            """
            
            params = {
                'document_uid': self.document_uid,
                'user_uid': user_uid,
                'assigned_by': assigned_by,
                'assigned_date': assigned_date.isoformat(),
                'expires_date': expires_date.isoformat(),
                'validity_days': validity_days
            }
            
            logger.info(f"Executing training assignment query with params: {params}")
            
            result = db_operations.run_query(query, params)
            
            if result and len(list(result)) > 0:
                logger.info(f"Training assignment successful for user {user_uid}")
                return True
            else:
                logger.error(f"Training assignment failed for user {user_uid} - no result returned")
                return False
            
        except Exception as e:
            logger.error(f"Error assigning training to user {user_uid}: {e}")
            import traceback
            logger.error(f"Traceback: {traceback.format_exc()}")
            return False
    
    def remove_user_training(self, user_uid: str) -> bool:
        """Remove training requirement from a user."""
        try:
            # FIXED: Use REQUIRES_TRAINING relationship instead of TRAINED
            query = """
            MATCH (u:User {UID: $user_uid})-[r:REQUIRES_TRAINING]->(d:ControlledDocument {UID: $doc_uid})
            DELETE r
            RETURN count(r) as deleted
            """
            
            result = db.run_query(query, {
                "user_uid": user_uid,
                "doc_uid": self.document_uid
            })
            
            return bool(result and result[0].get("deleted", 0) > 0)
            
        except Exception as e:
            logger.error(f"Error removing training from user {user_uid}: {e}")
            return False
    
    def get_assigned_users(self) -> List[Dict[str, Any]]:
        """Get all users assigned to this document's training."""
        try:
            from CDocs.db import db_operations
            
            logger.info(f"Getting assigned users for document {self.document_uid}")
            
            # FIXED: Use ControlledDocument label consistently
            query = """
            MATCH (d:ControlledDocument {UID: $document_uid})<-[r:REQUIRES_TRAINING]-(u:User)
            RETURN u.UID as user_uid, 
                   u.name as user_name, 
                   u.Mail as user_email,
                   r.status as status,
                   r.assigned_date as assigned_date,
                   r.trained_on as trained_on,
                   r.expires_date as expires_date,
                   r.assigned_by as assigned_by,
                   r.validity_days as validity_days
            ORDER BY r.assigned_date DESC
            """
            
            result = db_operations.run_query(query, {'document_uid': self.document_uid})
            
            assigned_users = []
            if result:
                for record in result:
                    user_data = dict(record)
                    logger.info(f"Found assigned user: {user_data}")
                    assigned_users.append(user_data)
            
            logger.info(f"Found {len(assigned_users)} assigned users for document {self.document_uid}")
            return assigned_users
            
        except Exception as e:
            logger.error(f"Error getting assigned users: {e}")
            import traceback
            logger.error(f"Traceback: {traceback.format_exc()}")
            return []
    
    def reset_all_training(self) -> bool:
        """Reset all training statuses to REQUIRED (used when document is republished)."""
        try:
            # FIXED: Use REQUIRES_TRAINING relationship instead of TRAINED
            query = """
            MATCH (u:User)-[r:REQUIRES_TRAINING]->(d:ControlledDocument {UID: $doc_uid})
            SET r.status = 'REQUIRED',
                r.trained_on = null,
                r.expires_date = null,
                r.reset_date = datetime()
            RETURN count(r) as reset_count
            """
        
            result = db.run_query(query, {"doc_uid": self.document_uid})
        
            return bool(result and result[0].get("reset_count", 0) >= 0)
        
        except Exception as e:
            logger.error(f"Error resetting training for document {self.document_uid}: {e}")
            return False

Parameters

Parameter Details

document_uid: Unique identifier for the document. If provided alone, the class will load existing training data from the database. Can be None if data parameter is provided.

data: Dictionary containing pre-loaded training data. If provided, bypasses database loading. Expected keys: 'document_uid', 'training_required', 'validity_days', 'quiz_required', 'instructions'. Can be None if document_uid is provided.

Return Value

Instantiation returns a DocumentTraining object that represents training requirements for a specific document. Methods return boolean values (True/False) for success/failure of operations, or lists of dictionaries for query operations like get_assigned_users() which returns user assignment data including user_uid, user_name, user_email, status, assigned_date, trained_on, expires_date, assigned_by, and validity_days.

Class Interface

Methods

Attributes

Dependencies

• typing
• datetime
• logging
• traceback
• CDocs.db
• CDocs.db.db_operations
• CDocs.models.BaseModel
• CDocs.models.register_model
• CDocs.models.user_extensions.DocUser
• CDocs.config.settings

Required Imports

from typing import Dict, List, Any, Optional
from datetime import datetime, timedelta
import logging
from CDocs.models import BaseModel
from CDocs.db import db_operations as db

Conditional/Optional Imports

These imports are only needed under specific conditions:

from datetime import datetime, timedelta

from CDocs.db import db_operations

import traceback

Usage Example

# Initialize with document UID to load existing training data
training = DocumentTraining(document_uid='DOC-123')

# Or initialize with pre-loaded data
data = {'document_uid': 'DOC-123', 'training_required': True, 'validity_days': 365}
training = DocumentTraining(data=data)

# Enable training for a document
success = training.enable_training(validity_days=365, quiz_required=True, instructions='Complete the quiz')

# Assign training to a user
assigned = training.assign_user_training(user_uid='USER-456', assigned_by='ADMIN-789', validity_days=365)

# Get all users assigned to this training
users = training.get_assigned_users()
for user in users:
    print(f"User: {user['user_name']}, Status: {user['status']}")

# Remove training from a user
removed = training.remove_user_training(user_uid='USER-456')

# Disable training for the document
disabled = training.disable_training()

# Reset all training statuses (e.g., after document republish)
reset = training.reset_all_training()

Best Practices

• Always provide either document_uid or data parameter during initialization, not both
• The class has a duplicate enable_training method - one instance method and one static method. Use the instance method for object-based operations and the static method for standalone document enabling
• Call enable_training() before assigning training to users to ensure the document has training enabled
• Use assign_user_training() to create REQUIRES_TRAINING relationships between users and documents
• The validity_days parameter determines how long training remains valid before expiring
• Call reset_all_training() when republishing a document to invalidate all previous training completions
• Always check return values (boolean) to verify operation success before proceeding
• The class maintains internal state in self._data which is synchronized with database operations
• Database operations use Neo4j Cypher queries and require proper database connection setup
• Error handling logs errors but returns False on failure - check logs for detailed error information
• The class uses ControlledDocument label in Neo4j - ensure your database schema matches this
• Training assignments create REQUIRES_TRAINING relationships with properties: status, assigned_date, expires_date, validity_days, trained_on
• get_assigned_users() returns a list that may be empty if no users are assigned

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class UserTraining 89.8% 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

• function assign_user_training 75.7% similar

  Assigns training requirements to multiple users for a specific controlled document, validating permissions, document training status, and user existence before creating assignments.

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

• function get_document_training_info 75.6% 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 enable_document_training 71.1% similar

  Enables training requirements for a controlled document, setting validity period, quiz requirements, and instructions for users who need to complete training on the document.

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

• class TrainingManagement 70.9% similar

  UI component for managing document training.

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