🔍 Code Extractor

Retrieves audit events from a Neo4j graph database with flexible filtering options including resource, user, event type, category, and date range filters, with pagination support.

Purpose

This function queries a Neo4j database to fetch audit trail events with comprehensive filtering capabilities. It's designed for audit logging systems where you need to track user actions, resource changes, and system events. The function supports pagination for large result sets and automatically enriches events with user names. It's particularly useful for compliance reporting, security monitoring, and debugging user activities.

Source Code

def get_audit_events(resource_uid: Optional[str] = None,
                    event_types: Optional[List[str]] = None,
                    event_category: Optional[str] = None,
                    user_uid: Optional[str] = None,
                    start_date: Optional[datetime] = None,
                    end_date: Optional[datetime] = None,
                    limit: int = 100,
                    skip: int = 0) -> List[Dict[str, Any]]:
    """
    Get audit events with optional filtering.
    
    Args:
        resource_uid: Optional UID of resource to get events for
        event_types: Optional list of event types to filter by
        event_category: Optional event category to filter by
        user_uid: Optional UID of user to get events for
        start_date: Optional start date for events
        end_date: Optional end date for events
        limit: Maximum number of events to return
        skip: Number of events to skip (for pagination)
        
    Returns:
        List of audit events
    """
    try:
        # Start building query
        query = "MATCH (e:AuditEvent) "
        conditions = []
        params = {}
        
        # Add resource filter if provided
        if resource_uid:
            query += "MATCH (e)-[:REFERS_TO]->(r {UID: $resource_uid}) "
            params['resource_uid'] = resource_uid
            
        # Add user filter if provided
        if user_uid:
            query += "MATCH (e)-[:PERFORMED_BY]->(u:User {UID: $user_uid}) "
            params['user_uid'] = user_uid
            
        # Add event type filter if provided
        if event_types:
            conditions.append("e.eventType IN $event_types")
            params['event_types'] = event_types
            
        # Add event category filter if provided
        if event_category:
            conditions.append("e.eventCategory = $event_category")
            params['event_category'] = event_category
            
        # Add date range filters if provided
        if start_date:
            conditions.append("e.timestamp >= $start_date")
            params['start_date'] = start_date
            
        if end_date:
            conditions.append("e.timestamp <= $end_date")
            params['end_date'] = end_date
            
        # Add conditions to query if any
        if conditions:
            query += "WHERE " + " AND ".join(conditions) + " "
            
        # Add sorting, pagination and return
        query += f"""
            RETURN e, u.Name as userName
            ORDER BY e.timestamp DESC
            SKIP {skip} LIMIT {limit}
        """
        
        # Execute query
        result = db.run_query(query, params)
        
        # Convert to list of dictionaries
        events = []
        for record in result:
            event = record['e']
            event['userName'] = record.get('userName')
            
            # Convert details from string to dictionary if possible
            if 'details' in event and isinstance(event['details'], str):
                try:
                    import json
                    event['details'] = json.loads(event['details'])
                except:
                    pass
                    
            events.append(event)
            
        return events
        
    except Exception as e:
        logger.error(f"Error getting audit events: {e}")
        return []

Parameters

Parameter Details

resource_uid: Optional unique identifier (UID) of a specific resource to filter events. When provided, only returns events related to that resource through the REFERS_TO relationship in the graph database.

event_types: Optional list of event type strings to filter by (e.g., ['CREATE', 'UPDATE', 'DELETE']). Only events matching one of these types will be returned.

event_category: Optional string representing a category of events to filter by (e.g., 'SECURITY', 'DATA_ACCESS'). Only events in this category will be returned.

user_uid: Optional unique identifier of a user to filter events. When provided, only returns events performed by that specific user through the PERFORMED_BY relationship.

start_date: Optional datetime object representing the earliest timestamp for events to include. Events with timestamps before this date are excluded.

end_date: Optional datetime object representing the latest timestamp for events to include. Events with timestamps after this date are excluded.

limit: Maximum number of audit events to return in a single query. Defaults to 100. Used for pagination and controlling response size.

skip: Number of events to skip from the beginning of the result set. Defaults to 0. Used for pagination to retrieve subsequent pages of results.

Return Value

Type: List[Dict[str, Any]]

Returns a list of dictionaries, where each dictionary represents an audit event with properties from the Neo4j AuditEvent node. Each event includes all node properties plus an enriched 'userName' field. The 'details' field is automatically parsed from JSON string to dictionary if possible. Returns an empty list if an error occurs or no events match the filters. Events are sorted by timestamp in descending order (newest first).

Dependencies

• logging
• uuid
• typing
• datetime
• json
• CDocs

Required Imports

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

Conditional/Optional Imports

These imports are only needed under specific conditions:

import json

Usage Example

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

# Get all recent audit events (last 7 days)
start = datetime.now() - timedelta(days=7)
events = get_audit_events(start_date=start, limit=50)

# Get events for a specific resource
resource_events = get_audit_events(
    resource_uid='resource-123',
    limit=20
)

# Get security-related events by a specific user
user_security_events = get_audit_events(
    user_uid='user-456',
    event_category='SECURITY',
    event_types=['LOGIN', 'LOGOUT', 'PERMISSION_CHANGE'],
    start_date=datetime(2024, 1, 1),
    end_date=datetime(2024, 12, 31)
)

# Pagination example - get second page of results
page_2_events = get_audit_events(
    limit=100,
    skip=100
)

# Process results
for event in events:
    print(f"Event: {event.get('eventType')} by {event.get('userName')} at {event.get('timestamp')}")
    if 'details' in event:
        print(f"Details: {event['details']}")

Best Practices

• Always use pagination (limit and skip) when querying large audit logs to avoid memory issues and timeouts
• Provide specific filters (resource_uid, user_uid, date ranges) to narrow results and improve query performance
• The function returns an empty list on errors rather than raising exceptions, so check the logs if you get unexpected empty results
• Date filters use inclusive comparisons (>= for start_date, <= for end_date)
• The 'details' field is automatically parsed from JSON, but parsing failures are silently ignored to prevent data loss
• Results are always sorted by timestamp in descending order (newest first), plan your pagination accordingly
• Ensure proper indexes exist on AuditEvent.timestamp, AuditEvent.eventType, and AuditEvent.eventCategory for optimal performance
• The function depends on specific Neo4j graph relationships (REFERS_TO, PERFORMED_BY) being properly established
• Consider implementing caching for frequently accessed audit queries to reduce database load

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function count_audit_events 90.1% similar

  Counts audit events in a Neo4j graph database with flexible filtering options including resource, user, event type, category, and date range.

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

• function log_event 70.4% similar

  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.

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

• function get_user_activity 70.2% similar

  Retrieves user activity logs from a Neo4j graph database by querying AuditEvent nodes with flexible filtering options for date ranges, users, and action types.

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

• function get_audit_trail 69.8% similar

  Retrieves audit trail events from a Neo4j database with optional filtering by date range, action type, and user, returning a list of audit event dictionaries.

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

• function get_document_history 61.7% similar

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

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