🔍 Code Extractor

function get_document_history

Maturity: 55

Retrieves the complete audit history for a document by querying events across document, version, review, comment, and approval nodes in a Neo4j graph database.

File:
/tf/active/vicechatdev/CDocs/utils/audit_trail.py
Lines:
593 - 667
Complexity:
complex

Purpose

This function provides comprehensive audit trail functionality for document management systems. It aggregates all events related to a specific document including direct document events, version changes, review cycles, comments, and approvals. The events are collected from multiple related nodes in the graph database and returned in chronological order, enabling full traceability and compliance tracking for controlled documents.

Source Code

def get_document_history(document_uid: str) -> List[Dict[str, Any]]:
    """
    Get complete history of events for a document.
    
    Args:
        document_uid: UID of document
        
    Returns:
        List of audit events in chronological order
    """
    try:
        all_events = []
        
        # Query 1: Get document events
        query_document_events = """
        MATCH (t:AuditTrail)-[:HAS_EVENT]->(e:AuditEvent)-[:REFERS_TO]->(d:ControlledDocument {UID: $doc_uid})
        OPTIONAL MATCH (e)-[:PERFORMED_BY]->(u:User)
        RETURN e, u.Name as userName
        """
        result = db.run_query(query_document_events, {"doc_uid": document_uid})
        all_events.extend([_process_event_record(record) for record in result])
        
        # Query 2: Get version events
        query_version_events = """
        MATCH (t:AuditTrail)-[:HAS_EVENT]->(e:AuditEvent)-[:REFERS_TO]->(v:DocumentVersion)
        MATCH (d:ControlledDocument {UID: $doc_uid})-[:HAS_VERSION]->(v)
        OPTIONAL MATCH (e)-[:PERFORMED_BY]->(u:User)
        RETURN e, u.Name as userName
        """
        result = db.run_query(query_version_events, {"doc_uid": document_uid})
        all_events.extend([_process_event_record(record) for record in result])
        
        # Query 3: Get review events
        query_review_events = """
        MATCH (d:ControlledDocument {UID: $doc_uid})-[:HAS_VERSION]->(v:DocumentVersion)
        MATCH (v)-[:FOR_REVIEW]->(r:ReviewCycle)
        MATCH (t:AuditTrail)-[:HAS_EVENT]->(e:AuditEvent)-[:REFERS_TO]->(r)
        OPTIONAL MATCH (e)-[:PERFORMED_BY]->(u:User)
        RETURN e, u.Name as userName
        """
        result = db.run_query(query_review_events, {"doc_uid": document_uid})
        all_events.extend([_process_event_record(record) for record in result])
        
        # Query 4: Get comment events
        query_comment_events = """
        MATCH (d:ControlledDocument {UID: $doc_uid})-[:HAS_VERSION]->(v:DocumentVersion)
        MATCH (v)-[:FOR_REVIEW]->(r:ReviewCycle)
        MATCH (r)-[:COMMENTED_ON]->(c:ReviewComment)
        MATCH (t:AuditTrail)-[:HAS_EVENT]->(e:AuditEvent)-[:REFERS_TO]->(c)
        OPTIONAL MATCH (e)-[:PERFORMED_BY]->(u:User)
        RETURN e, u.Name as userName
        """
        result = db.run_query(query_comment_events, {"doc_uid": document_uid})
        all_events.extend([_process_event_record(record) for record in result])
        
        # Query 5: Get approval events
        query_approval_events = """
        MATCH (d:ControlledDocument {UID: $doc_uid})-[:HAS_VERSION]->(v:DocumentVersion)
        MATCH (v)-[:FOR_REVIEW]->(r:ReviewCycle)
        MATCH (r)-[:FOR_APPROVAL]->(a:Approval)
        MATCH (t:AuditTrail)-[:HAS_EVENT]->(e:AuditEvent)-[:REFERS_TO]->(a)
        OPTIONAL MATCH (e)-[:PERFORMED_BY]->(u:User)
        RETURN e, u.Name as userName
        """
        result = db.run_query(query_approval_events, {"doc_uid": document_uid})
        all_events.extend([_process_event_record(record) for record in result])
        
        # Sort all events by timestamp
        all_events.sort(key=lambda event: event.get('timestamp', datetime.min))
        
        return all_events
        
    except Exception as e:
        logger.error(f"Error getting document history: {e}")
        return []

Parameters

Name Type Default Kind
document_uid str - positional_or_keyword

Parameter Details

document_uid: Unique identifier (UID) string for the controlled document. This should be a valid UID that exists in the Neo4j database as a ControlledDocument node. Used to query all related audit events across the document lifecycle.

Return Value

Type: List[Dict[str, Any]]

Returns a List of dictionaries where each dictionary represents an audit event. Each event dictionary contains processed event data including timestamp, event type, user information, and other event-specific details. Events are sorted chronologically by timestamp. Returns an empty list if an error occurs or if no events are found for the document.

Dependencies

  • logging
  • uuid
  • typing
  • datetime
  • json
  • CDocs.db
  • CDocs.config
  • CDocs.db.schema_manager
  • CDocs.models.user_extensions

Required Imports

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

Usage Example

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

# Assuming db connection and _process_event_record are configured

# Get complete history for a document
document_uid = "DOC-12345-ABCD"
history = get_document_history(document_uid)

# Process the returned events
if history:
    print(f"Found {len(history)} events for document {document_uid}")
    for event in history:
        timestamp = event.get('timestamp', 'Unknown')
        event_type = event.get('eventType', 'Unknown')
        user = event.get('userName', 'System')
        print(f"{timestamp}: {event_type} by {user}")
else:
    print("No history found or error occurred")

# Example output structure:
# [
#   {
#     'timestamp': datetime(2024, 1, 15, 10, 30),
#     'eventType': 'DOCUMENT_CREATED',
#     'userName': 'John Doe',
#     'details': {...}
#   },
#   {
#     'timestamp': datetime(2024, 1, 16, 14, 20),
#     'eventType': 'VERSION_CREATED',
#     'userName': 'Jane Smith',
#     'details': {...}
#   }
# ]

Best Practices

  • Ensure the document_uid exists in the database before calling this function to avoid empty results
  • The function returns an empty list on errors, so check the logs for detailed error information
  • The function depends on _process_event_record helper function which must be implemented in the same module
  • Performance may degrade with documents that have extensive histories; consider pagination for large result sets
  • The function executes 5 separate database queries which could be optimized with a single complex query if performance is critical
  • Ensure proper database indexes on UID fields and relationship types for optimal query performance
  • The chronological sorting uses datetime.min as fallback for events without timestamps, which places them at the beginning of the timeline
  • Error handling catches all exceptions and logs them, but returns empty list - consider if different error handling is needed for your use case

Tags

audit-trail document-management neo4j graph-database history-tracking compliance event-logging document-lifecycle version-control review-management

Similar Components

AI-powered semantic similarity - components with related functionality:

  • function get_document_audit_trail 83.7% similar

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

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

  • function generate_audit_report 77.9% similar

    Generates a comprehensive audit report for a controlled document by aggregating document metadata, version history, review cycles, approvals, and categorized audit events.

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

  • function get_document_with_relationships 71.2% similar

    Retrieves a complete document from Neo4j graph database along with all its related entities including versions, reviews, approvals, and authors.

    From: /tf/active/vicechatdev/CDocs/db/db_operations.py

  • function get_document 70.0% 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 get_document_v1 67.8% similar

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

    From: /tf/active/vicechatdev/CDocs/controllers/document_controller.py
← Back to Browse