🔍 Code Extractor

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

Purpose

UserTraining manages the lifecycle of training requirements for users on controlled documents. It handles loading training data from a Neo4j graph database via REQUIRES_TRAINING relationships, marking training as complete with expiration dates, calculating effective training status (including expired training), and retrieving all training requirements for a user. This class is essential for compliance tracking in document management systems where users must be trained on specific controlled documents.

Source Code

class UserTraining(BaseModel):
    """
    Model representing a user's training status for a specific document.
    """
    
    def __init__(self, user_uid: str, document_uid: str, data: Dict[str, Any] = None):
        """Initialize UserTraining instance."""
        self.user_uid = user_uid
        self.document_uid = document_uid
        if data:
            super().__init__(data)
        else:
            # Load training data for user and document
            training_data = self._load_training_data()
            super().__init__(training_data or {})
    
    def _load_training_data(self) -> Optional[Dict[str, Any]]:
        """Load training data for user and document."""
        if not self.user_uid or not self.document_uid:
            return None
            
        # FIXED: Use REQUIRES_TRAINING relationship
        query = """
        MATCH (u:User {UID: $user_uid})-[r:REQUIRES_TRAINING]->(d:ControlledDocument {UID: $doc_uid})
        RETURN u.name as user_name,
               u.Mail as user_email,
               d.doc_number as doc_number,
               d.title as doc_title,
               r.assigned_date as assigned_date,
               r.status as status,
               r.trained_on as trained_on,
               r.expires_date as expires_on,
               r.assigned_by as assigned_by
        """
        
        result = db.run_query(query, {
            "user_uid": self.user_uid,
            "doc_uid": self.document_uid
        })
        
        if result:
            row = result[0]
            return {
                "user_uid": self.user_uid,
                "document_uid": self.document_uid,
                "user_name": row.get("user_name"),
                "user_email": row.get("user_email"),
                "doc_number": row.get("doc_number"),
                "doc_title": row.get("doc_title"),
                "assigned_date": row.get("assigned_date"),
                "status": row.get("status", "REQUIRED"),
                "trained_on": row.get("trained_on"),
                "expires_on": row.get("expires_on"),
                "assigned_by": row.get("assigned_by")
            }
        return None
    
    def mark_trained(self, validity_days: int = 365) -> bool:
        """Mark user training as complete."""
        try:
            from datetime import datetime, timedelta
            
            trained_date = datetime.now()
            expires_date = trained_date + timedelta(days=validity_days)
            
            # FIXED: Use REQUIRES_TRAINING relationship and update to TRAINED status
            query = """
            MATCH (u:User {UID: $user_uid})-[r:REQUIRES_TRAINING]->(d:ControlledDocument {UID: $doc_uid})
            SET r.status = 'TRAINED',
                r.trained_on = $trained_date,
                r.expires_date = $expires_date,
                r.completion_date = datetime()
            RETURN r
            """
            
            result = db.run_query(query, {
                "user_uid": self.user_uid,
                "doc_uid": self.document_uid,
                "trained_date": trained_date.isoformat(),
                "expires_date": expires_date.isoformat()
            })
            
            if result:
                self._data.update({
                    "status": "TRAINED",
                    "trained_on": trained_date,
                    "expires_on": expires_date
                })
                return True
            return False
            
        except Exception as e:
            logger.error(f"Error marking training as complete: {e}")
            return False
    
    def get_effective_status(self) -> str:
        """Get the effective training status (considering expiration)."""
        status = self._data.get("status", "REQUIRED")
        
        if status == "TRAINED":
            expires_on = self._data.get("expires_on")
            if expires_on and expires_on < datetime.now():
                return "EXPIRED"
        
        return status
    
    @classmethod
    def get_user_training_requirements(cls, user_uid: str) -> List[Dict[str, Any]]:
        """Get all training requirements for a user."""
        try:
            # FIXED: Look for REQUIRES_TRAINING relationships instead of TRAINED
            query = """
            MATCH (u:User {UID: $user_uid})-[r:REQUIRES_TRAINING]->(d:ControlledDocument)
            WHERE d.training_required = true
            RETURN u.UID as user_uid,
                   u.name as user_name,
                   d.UID as document_uid,
                   d.doc_number as doc_number,
                   d.title as doc_title,
                   d.training_validity_days as validity_days,
                   d.training_quiz_required as quiz_required,
                   d.training_instructions as instructions,
                   r.assigned_date as assigned_date,
                   r.status as status,
                   r.trained_on as trained_on,
                   r.expires_date as expires_date
            ORDER BY r.assigned_date DESC
            """
            
            result = db.run_query(query, {"user_uid": user_uid})
            
            training_reqs = []
            if result:  # Add this check
                for row in result:
                    # Calculate effective status
                    status = row.get("status", "REQUIRED")
                    if status == "TRAINED":
                        expires_date = row.get("expires_date")
                        if expires_date:
                            try:
                                if isinstance(expires_date, str):
                                    expires_dt = datetime.fromisoformat(expires_date.replace('Z', '+00:00'))
                                else:
                                    expires_dt = expires_date
                                
                                if expires_dt < datetime.now():
                                    status = "EXPIRED"
                            except Exception as e:
                                logger.warning(f"Error parsing expires_date: {e}")
                    
                    training_reqs.append({
                        "user_uid": row.get("user_uid"),
                        "user_name": row.get("user_name"),
                        "document_uid": row.get("document_uid"),
                        "doc_number": row.get("doc_number"),
                        "doc_title": row.get("doc_title"),
                        "validity_days": row.get("validity_days", 365),
                        "quiz_required": row.get("quiz_required", False),
                        "instructions": row.get("instructions", ""),
                        "assigned_date": row.get("assigned_date"),
                        "status": row.get("status"),
                        "effective_status": status,
                        "trained_on": row.get("trained_on"),
                        "expires_on": row.get("expires_date")  # Map expires_date to expires_on for consistency
                    })
            
            logger.info(f"Found {len(training_reqs)} training requirements for user {user_uid}")
            return training_reqs
            
        except Exception as e:
            logger.error(f"Error getting user training requirements: {e}")
            import traceback
            logger.error(f"Traceback: {traceback.format_exc()}")
            return []

Parameters

Parameter Details

user_uid: Unique identifier (UID) for the user. Required to identify which user's training status is being tracked. Must correspond to a User node in the Neo4j database.

document_uid: Unique identifier (UID) for the controlled document. Required to identify which document the training is for. Must correspond to a ControlledDocument node in the Neo4j database.

data: Optional dictionary containing pre-loaded training data. If provided, bypasses database loading and initializes the instance with this data. If None, training data is loaded from the database via _load_training_data(). Expected keys include: user_name, user_email, doc_number, doc_title, assigned_date, status, trained_on, expires_on, assigned_by.

Return Value

Instantiation returns a UserTraining object that inherits from BaseModel. The object contains training status data accessible via the _data attribute. Key methods return: mark_trained() returns bool (True if successfully marked as trained), get_effective_status() returns str (status considering expiration: 'REQUIRED', 'TRAINED', or 'EXPIRED'), get_user_training_requirements() returns List[Dict[str, Any]] containing all training requirements for a user.

Class Interface

Methods

Attributes

Dependencies

• typing
• datetime
• logging
• traceback
• CDocs.db.db_operations
• CDocs.models.BaseModel

Required Imports

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

Conditional/Optional Imports

These imports are only needed under specific conditions:

import traceback

from datetime import datetime, timedelta

Usage Example

# Instantiate with database loading
training = UserTraining(user_uid='user123', document_uid='doc456')

# Check current status
status = training.get_effective_status()
print(f"Training status: {status}")

# Mark training as complete with 365-day validity
if training.mark_trained(validity_days=365):
    print("Training marked as complete")

# Instantiate with pre-loaded data
data = {
    'user_uid': 'user123',
    'document_uid': 'doc456',
    'status': 'REQUIRED',
    'assigned_date': '2024-01-01'
}
training2 = UserTraining(user_uid='user123', document_uid='doc456', data=data)

# Get all training requirements for a user
requirements = UserTraining.get_user_training_requirements('user123')
for req in requirements:
    print(f"{req['doc_number']}: {req['effective_status']}")

Best Practices

• Always provide valid user_uid and document_uid that exist in the database to avoid None returns from _load_training_data()
• Use get_effective_status() instead of directly accessing status to account for expired training
• Call mark_trained() only after user has completed training; it updates the database immediately
• The validity_days parameter in mark_trained() defaults to 365 but should match document-specific requirements
• Use the class method get_user_training_requirements() to retrieve all requirements for a user rather than instantiating multiple objects
• Handle the case where _load_training_data() returns None (no training relationship exists)
• The class modifies database state via mark_trained(); ensure proper error handling
• Status values are: 'REQUIRED' (not yet trained), 'TRAINED' (completed and valid), 'EXPIRED' (completed but past expiration)
• The class inherits from BaseModel, so training data is stored in self._data dictionary
• Date fields (trained_on, expires_on) are stored as ISO format strings in the database

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class DocumentTraining 89.8% similar

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

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

• function get_document_training_info 73.2% 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 complete_user_training_by_uid 72.7% 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

• function assign_user_training 72.0% 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 complete_user_training 70.5% similar

  Marks a user's training as complete for a specific controlled document, validating quiz requirements and setting expiration dates.

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