🔍 Code Extractor

function get_audit_trail

Maturity: 55

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.

File:
/tf/active/vicechatdev/CDocs/controllers/admin_controller.py
Lines:
629 - 688
Complexity:
moderate

Purpose

This function queries the database for audit events to display in an admin panel. It supports flexible filtering by date range (date_from, date_to), specific action types, and username patterns. The function builds dynamic Cypher queries based on provided filters, retrieves audit events with associated user information, and returns formatted audit records sorted by timestamp in descending order. It's designed for administrative monitoring and compliance tracking of system activities.

Source Code

def get_audit_trail(limit: int = 100, filters: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
    """Get audit trail for admin panel."""
    try:
        # Build query conditions
        conditions = []
        params = {'limit': limit}
        
        if filters:
            if filters.get('date_from'):
                conditions.append("a.timestamp >= $date_from")
                params['date_from'] = filters['date_from']
                
            if filters.get('date_to'):
                conditions.append("a.timestamp <= $date_to")
                params['date_to'] = filters['date_to']
                
            if filters.get('action'):
                conditions.append("a.action = $action")
                params['action'] = filters['action']
                
            if filters.get('user'):
                conditions.append("u.username CONTAINS $user")
                params['user'] = filters['user']
        
        where_clause = ""
        if conditions:
            where_clause = "WHERE " + " AND ".join(conditions)
        
        query = f"""
        MATCH (a:AuditEvent)
        OPTIONAL MATCH (a)-[:PERFORMED_BY]->(u:DocUser)
        {where_clause}
        RETURN a.timestamp as timestamp,
               u.username as user,
               a.action as action,
               a.resource as resource,
               a.details as details,
               a.ipAddress as ip_address
        ORDER BY a.timestamp DESC
        LIMIT $limit
        """
        
        results = db.run_query(query, params)
        audit_events = []
        
        for record in results:
            audit_events.append({
                'timestamp': record.get('timestamp', ''),
                'user': record.get('user', 'System'),
                'action': record.get('action', ''),
                'resource': record.get('resource', ''),
                'details': record.get('details', ''),
                'ip_address': record.get('ip_address', '')
            })
        
        return audit_events
        
    except Exception as e:
        logger.error(f"Error getting audit trail: {e}")
        return []

Parameters

Name Type Default Kind
limit int 100 positional_or_keyword
filters Optional[Dict[str, Any]] None positional_or_keyword

Parameter Details

limit: Maximum number of audit events to return. Defaults to 100. Controls the result set size to prevent overwhelming responses. Must be a positive integer.

filters: Optional dictionary containing filter criteria. Supported keys: 'date_from' (timestamp for start of date range), 'date_to' (timestamp for end of date range), 'action' (exact action type to match), 'user' (username pattern to search with CONTAINS operator). If None or empty, no filters are applied and all audit events are returned up to the limit.

Return Value

Type: List[Dict[str, Any]]

Returns a list of dictionaries, where each dictionary represents an audit event with keys: 'timestamp' (when the event occurred), 'user' (username who performed the action, defaults to 'System' if no user), 'action' (type of action performed), 'resource' (resource affected by the action), 'details' (additional event details), 'ip_address' (IP address from which the action was performed). Returns an empty list if an error occurs or no events are found.

Dependencies

  • logging
  • typing
  • CDocs.db
  • CDocs.models.document
  • CDocs.models.user_extensions
  • CDocs.db.schema_manager
  • CDocs.config
  • CDocs.utils.audit_trail

Required Imports

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

Usage Example

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

logger = logging.getLogger(__name__)

# Get last 50 audit events without filters
audit_events = get_audit_trail(limit=50)

# Get audit events with date range filter
filters = {
    'date_from': '2024-01-01T00:00:00',
    'date_to': '2024-12-31T23:59:59'
}
audit_events = get_audit_trail(limit=100, filters=filters)

# Get audit events for specific action and user
filters = {
    'action': 'document_updated',
    'user': 'admin'
}
audit_events = get_audit_trail(limit=200, filters=filters)

# Process results
for event in audit_events:
    print(f"{event['timestamp']}: {event['user']} performed {event['action']} on {event['resource']}")

Best Practices

  • Always handle the returned list defensively as it may be empty on errors
  • Use reasonable limit values to avoid performance issues with large audit trails
  • Ensure date filters are in the correct format expected by the database timestamp fields
  • The function catches all exceptions and returns an empty list, so check logs for error details
  • The 'user' filter uses CONTAINS operator, so partial username matches will be returned
  • Results are always sorted by timestamp in descending order (newest first)
  • The function uses parameterized queries to prevent SQL injection attacks
  • Consider implementing pagination for large audit trails instead of relying solely on limit parameter

Tags

audit logging admin database neo4j cypher filtering compliance monitoring security query-builder event-tracking

Similar Components

AI-powered semantic similarity - components with related functionality:

  • function get_user_activity 79.0% 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_events 69.8% similar

    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.

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

  • function log_user_action 68.1% similar

    Creates an audit event node in a graph database to log user actions, connecting it to both an audit trail and the user who performed the action.

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

  • function count_audit_events 67.2% 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 get_document_audit_trail 65.2% 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
← Back to Browse