🔍 Code Extractor

Creates and persists an audit trail event in a Neo4j graph database, linking it to users, resources, and an audit trail node with comprehensive event metadata.

Purpose

This function serves as the primary mechanism for logging audit events in a document management system. It validates event types, creates AuditEvent nodes in Neo4j, establishes relationships with users and resources, and maintains a complete audit trail for compliance and tracking purposes. The function handles event categorization, timestamping, and flexible detail storage while ensuring data integrity through validation and error handling.

Source Code

def log_event(event_type: str, 
             user: Optional[Union[DocUser, str]], 
             resource_uid: Optional[str] = None,
             resource_type: Optional[str] = None,
             details: Optional[Dict[str, Any]] = None) -> Optional[str]:
    """
    Log an audit trail event.
    
    Args:
        event_type: Type of event from EVENT_TYPES
        user: User who performed the action or their UID
        resource_uid: UID of resource being acted upon (document, version, etc.)
        resource_type: Type of resource
        details: Additional details about the event
        
    Returns:
        UID of the created audit event or None if creation failed
    """
    try:
        # Validate event type
        if event_type not in EVENT_TYPES:
            logger.error(f"Invalid event type: {event_type}")
            return None
            
        # Get user UID
        user_uid = user.uid if isinstance(user, DocUser) else user
        
        # Prepare event properties
        event_info = EVENT_TYPES[event_type]
        props = {
            'UID': str(uuid.uuid4()),
            'eventType': event_type,
            'eventCategory': event_info['category'],
            'description': event_info['description'],
            'timestamp': datetime.now(),
            'icon': event_info.get('icon', 'activity')
        }
        
        if user_uid:
            props['userUID'] = user_uid
            
        if resource_uid:
            props['resourceUID'] = resource_uid
            
        if resource_type:
            props['resourceType'] = resource_type
            
        if details:
            props['details'] = str(details)
            
        # Ensure audit trail node exists and get its UID
        from CDocs.db import get_driver
        driver = get_driver()
        from CDocs.db.schema_manager import ensure_audit_trail_exists
        audit_trail_uid = ensure_audit_trail_exists(driver)
        
        if not audit_trail_uid:
            logger.error("Failed to ensure audit trail node exists")
            return None
        
        # Create event node under AuditTrail instead of directly under CDocs
        db.run_query(
            """
            MATCH (t:AuditTrail {UID: $audit_trail_uid})
            CREATE (t)-[:HAS_EVENT]->(e:AuditEvent)
            SET e = $props
            RETURN e.UID as eventUID
            """,
            {"props": props, "audit_trail_uid": audit_trail_uid}
        )
        
        # Link to user if user UID provided
        if user_uid:
            db.run_query(
                """
                MATCH (e:AuditEvent {UID: $event_uid}), (u:User {UID: $user_uid})
                CREATE (e)-[:PERFORMED_BY]->(u)
                """,
                {"event_uid": props['UID'], "user_uid": user_uid}
            )
            
        # Link to resource if resource UID provided
        if resource_uid:
            db.run_query(
                """
                MATCH (e:AuditEvent {UID: $event_uid})
                MATCH (r {UID: $resource_uid})
                CREATE (e)-[:REFERS_TO]->(r)
                """,
                {"event_uid": props['UID'], "resource_uid": resource_uid}
            )
            
        return props['UID']
        
    except Exception as e:
        logger.error(f"Error logging audit event: {e}")
        return None

Parameters

Parameter Details

event_type: A string identifier for the type of event being logged. Must be a valid key from the EVENT_TYPES dictionary (not shown in code but referenced). Examples might include 'document_created', 'user_login', 'permission_changed', etc. Invalid event types will cause the function to return None.

user: The user who performed the action. Can be either a DocUser object (from which the UID will be extracted) or a string representing the user's UID directly. Can be None if the event is not associated with a specific user. If provided, a PERFORMED_BY relationship will be created in the database.

resource_uid: Optional string representing the unique identifier of the resource being acted upon (e.g., document UID, version UID, folder UID). If provided, a REFERS_TO relationship will be created linking the event to the resource node in the database.

resource_type: Optional string describing the type of resource being referenced (e.g., 'Document', 'Version', 'Folder'). This is stored as metadata on the event node for categorization and filtering purposes.

details: Optional dictionary containing additional contextual information about the event. Can include any key-value pairs relevant to the specific event type. The dictionary will be converted to a string for storage in the database.

Return Value

Type: Optional[str]

Returns a string containing the UUID of the newly created AuditEvent node if the operation succeeds. Returns None if the operation fails due to invalid event type, database errors, or failure to ensure the audit trail node exists. The returned UID can be used to reference or query the specific audit event later.

Dependencies

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

Required Imports

import logging
import uuid
from typing import Dict, List, Any, Optional, Union
from datetime import datetime
import json
from CDocs import db
from CDocs.config import settings
from CDocs.db.schema_manager import NodeLabels, RelTypes
from CDocs.models.user_extensions import DocUser

Conditional/Optional Imports

These imports are only needed under specific conditions:

from CDocs.db import get_driver

from CDocs.db.schema_manager import ensure_audit_trail_exists

Usage Example

# Assuming EVENT_TYPES is defined and database is configured
from CDocs.models.user_extensions import DocUser
from your_module import log_event

# Example 1: Log a document creation event with a DocUser object
user = DocUser(uid='user-123', name='John Doe')
event_uid = log_event(
    event_type='document_created',
    user=user,
    resource_uid='doc-456',
    resource_type='Document',
    details={'title': 'Project Proposal', 'size': 1024}
)
if event_uid:
    print(f'Event logged with UID: {event_uid}')

# Example 2: Log a user login event with just a user UID
event_uid = log_event(
    event_type='user_login',
    user='user-789',
    details={'ip_address': '192.168.1.1', 'browser': 'Chrome'}
)

# Example 3: Log a system event without a user
event_uid = log_event(
    event_type='system_backup',
    user=None,
    details={'backup_size': '5GB', 'duration': '30min'}
)

Best Practices

• Always ensure EVENT_TYPES dictionary is properly defined before calling this function
• Handle the None return value to detect and respond to logging failures
• Use DocUser objects when available for type safety, but string UIDs are acceptable for flexibility
• Keep the details dictionary serializable (avoid complex objects) since it will be converted to string
• Ensure the Neo4j database connection is established and healthy before calling this function
• Consider implementing retry logic for transient database failures in production environments
• Use meaningful event_type values that are pre-defined in EVENT_TYPES for consistency
• Always provide resource_uid and resource_type together when logging resource-related events
• Monitor the logger output for validation errors and database issues
• Ensure the database schema includes proper indexes on UID fields for performance
• The function creates multiple database queries sequentially; consider transaction management for atomicity in critical scenarios

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function log_version_event 77.4% similar

  A convenience wrapper function that logs audit events specifically for document version operations by delegating to a generic log_event function with the resource_type fixed as 'DocumentVersion'.

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

• function log_document_lifecycle_event 76.0% similar

  Logs a document lifecycle event to an audit trail by delegating to a generic log_event function with document-specific parameters.

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

• function log_review_event 75.1% similar

  A convenience wrapper function that logs audit events specifically for review cycle operations by delegating to a generic log_event function with the resource_type pre-set to 'ReviewCycle'.

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

• function log_training_event 73.1% similar

  A wrapper function that logs training-related events by delegating to a generic log_event function with pre-configured resource type.

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

• function log_approval_event 72.2% similar

  A convenience wrapper function that logs approval-related events by delegating to a generic log_event function with the resource_type fixed as 'Approval'.

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