🔍 Code Extractor

Retrieves the complete audit trail for a controlled document from a Neo4j graph database, including timestamps, user actions, and event details.

Purpose

This function queries a Neo4j database to fetch all audit events related to a specific controlled document. It returns a chronologically ordered list of audit entries showing who performed what actions on the document, when they occurred, and any associated comments or details. This is essential for compliance, tracking document history, and maintaining regulatory audit trails in document management systems.

Source Code

def get_document_audit_trail(document_uid: str) -> List[Dict[str, Any]]:
    """
    Get the audit trail for a document
    
    Parameters
    ----------
    document_uid : str
        ID of the document
        
    Returns
    -------
    List[Dict[str, Any]]
        Audit trail entries
    """
    try:
        # Query audit events related to the document
        query = """
        MATCH (d:ControlledDocument {UID: $document_uid})<-[:RELATES_TO]-(e:AuditEvent)
        RETURN e
        ORDER BY e.timestamp ASC
        """
        
        result = db.run_query(query, {"document_uid": document_uid})
        
        if not result:
            return []
            
        # Format the audit entries
        audit_entries = []
        for record in result:
            if 'e' in record:
                event = record['e']
                
                # Extract user details if available
                user_name = None
                if 'userUID' in event:
                    user_query = "MATCH (u:User {UID: $user_uid}) RETURN u.name as name"
                    user_result = db.run_query(user_query, {"user_uid": event['userUID']})
                    if user_result and 'name' in user_result[0]:
                        user_name = user_result[0]['name']
                
                # Format the event for the audit trail
                audit_entry = {
                    'timestamp': event.get('timestamp', ''),
                    'user': event.get('userUID', ''),
                    'user_name': user_name or 'Unknown User',
                    'action': event.get('eventType', ''),
                    'comment': event.get('message', ''),
                    'details': event.get('details', {})
                }
                
                audit_entries.append(audit_entry)
                
        return audit_entries
        
    except Exception as e:
        logger.error(f"Error retrieving audit trail for document {document_uid}: {str(e)}")
        return []

Parameters

Parameter Details

document_uid: Unique identifier (UID) string for the controlled document. This should be a valid document UID that exists in the Neo4j database as a ControlledDocument node. The function uses this to query related AuditEvent nodes.

Return Value

Type: List[Dict[str, Any]]

Returns a List of dictionaries, where each dictionary represents an audit trail entry with the following keys: 'timestamp' (when the event occurred), 'user' (user UID who performed the action), 'user_name' (resolved user name or 'Unknown User'), 'action' (type of event/action performed), 'comment' (message associated with the event), and 'details' (additional event details as a dictionary). Returns an empty list if no audit events are found or if an error occurs. The list is ordered chronologically by timestamp in ascending order.

Dependencies

• logging
• typing
• CDocs.db
• neo4j

Required Imports

from typing import List, Dict, Any
import logging
from CDocs import db

Usage Example

from typing import List, Dict, Any
from CDocs import db
import logging

# Ensure logger is configured
logger = logging.getLogger(__name__)

# Get audit trail for a specific document
document_uid = "doc-12345-abcde"
audit_trail = get_document_audit_trail(document_uid)

# Process the audit trail
if audit_trail:
    for entry in audit_trail:
        print(f"{entry['timestamp']}: {entry['user_name']} performed {entry['action']}")
        if entry['comment']:
            print(f"  Comment: {entry['comment']}")
        if entry['details']:
            print(f"  Details: {entry['details']}")
else:
    print("No audit trail found or error occurred")

Best Practices

• Always check if the returned list is empty before processing audit entries
• The function returns an empty list on errors rather than raising exceptions, so check logs for error details
• Ensure the document_uid exists in the database before calling this function to avoid unnecessary queries
• The function performs multiple database queries (one for events, then one per event for user details), which may impact performance for documents with many audit events
• Consider caching user name lookups if calling this function frequently to reduce database load
• The function gracefully handles missing user information by defaulting to 'Unknown User'
• Audit entries are returned in chronological order (oldest first) based on the timestamp field
• The 'details' field may contain nested dictionaries with additional event-specific information

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function get_document 67.3% similar

  Retrieves comprehensive details of a controlled document by its UID, with optional inclusion of version history, review cycles, and approval cycles.

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

• function create_audit_entry 66.0% similar

  Creates an audit trail entry for document actions, logging user activity, document changes, and lifecycle events with fallback mechanisms.

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

• function get_document_versions_v1 62.6% similar

  Retrieves all versions of a document from the database given its unique identifier (UID).

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

• function get_documents 58.9% similar

  Retrieves filtered and paginated documents from a Neo4j graph database with permission-based access control, supporting multiple filter criteria and search functionality.

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

• function download_document_version 57.8% similar

  Downloads a specific version of a controlled document from FileCloud storage, with optional audit trail and watermark inclusion, and logs the download event.

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