🔍 Code Extractor

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

Purpose

This function provides a flexible interface to query and retrieve audit trail data from a Neo4j database. It supports filtering by date range (ISO format), specific user IDs, action types, and result limits. The function constructs dynamic Cypher queries based on provided parameters, executes them against the database, and returns formatted activity records. It includes debugging capabilities to log database structure and query results, and gracefully handles errors by returning an empty list on failure.

Source Code

def get_user_activity(date_from=None, date_to=None, user_id=None, limit=100, action_types=None):
    """
    Get user activity logs from the database.
    
    Parameters
    ----------
    date_from : str, optional
        Start date for activity filtering (ISO format)
    date_to : str, optional
        End date for activity filtering (ISO format)
    user_id : str, optional
        Filter by specific user ID
    limit : int, optional
        Maximum number of records to return
    action_types : list, optional
        List of action types to filter by
        
    Returns
    -------
    list
        List of activity records with timestamp, user, action type and details
    """
    try:
        # Build query conditions
        conditions = []
        params = {'limit': limit}
        
        if date_from:
            conditions.append("a.timestamp >= $date_from")
            params['date_from'] = date_from
            
        if date_to:
            conditions.append("a.timestamp <= $date_to")
            params['date_to'] = date_to
            
        if user_id:
            conditions.append("a.userUID = $user_id")
            params['user_id'] = user_id
            
        if action_types:
            conditions.append("a.eventType IN $action_types")
            params['action_types'] = action_types
            
        # Construct where clause
        where_clause = ""
        if conditions:
            where_clause = "WHERE " + " AND ".join(conditions)
            
        # Query for activity logs - Updated to use AuditEvent nodes
        query = f"""
        MATCH (a:AuditEvent)-[:PERFORMED_BY]->(u:User)
        {where_clause}
        RETURN a.timestamp as timestamp, 
               COALESCE(u.username, u.Name, 'Unknown') as user_name,
               a.eventType as action_type,
               COALESCE(a.description, a.details, '') as action_detail
        ORDER BY a.timestamp DESC
        LIMIT $limit
        """
        
        # Execute query using db.run_query instead of db_operations.execute_read_query
        results = db.run_query(query, params)
        
        # Debug: Log what we found and check database structure
        logger.info(f"Found {len(results) if results else 0} audit events")
        
        # Debug: Check what's actually in the database
        debug_query = "MATCH (a:AuditEvent) RETURN count(a) as total_events"
        debug_results = db.run_query(debug_query)
        if debug_results:
            logger.info(f"Total AuditEvent nodes in database: {debug_results[0].get('total_events', 0)}")
        
        # Debug: Check for any Activity nodes (old structure)
        old_debug_query = "MATCH (a:Activity) RETURN count(a) as total_activities"
        old_debug_results = db.run_query(old_debug_query)
        if old_debug_results:
            logger.info(f"Total Activity nodes in database: {old_debug_results[0].get('total_activities', 0)}")
        
        # Format results
        activity_logs = []
        for record in results:
            activity_logs.append({
                'timestamp': record.get('timestamp'),
                'user_name': record.get('user_name'),
                'action_type': record.get('action_type'),
                'action_detail': record.get('action_detail')
            })
            
        return activity_logs
        
    except Exception as e:
        logger.error(f"Error getting user activity: {str(e)}")
        return []

Parameters

Parameter Details

date_from: Optional start date for filtering activity logs in ISO format (e.g., '2024-01-01T00:00:00'). Only activities on or after this date will be returned. If None, no lower date bound is applied.

date_to: Optional end date for filtering activity logs in ISO format (e.g., '2024-12-31T23:59:59'). Only activities on or before this date will be returned. If None, no upper date bound is applied.

user_id: Optional string identifier for filtering activities by a specific user. Should match the userUID property on AuditEvent nodes. If None, activities from all users are returned.

limit: Maximum number of activity records to return. Defaults to 100. Must be a positive integer. Results are ordered by timestamp descending (most recent first) before applying the limit.

action_types: Optional list of strings representing event types to filter by (e.g., ['login', 'document_update', 'user_created']). Should match the eventType property on AuditEvent nodes. If None, all action types are included.

Return Value

Returns a list of dictionaries, where each dictionary represents an activity log entry with keys: 'timestamp' (the event timestamp), 'user_name' (username or name of the user who performed the action, or 'Unknown' if not available), 'action_type' (the type of action performed), and 'action_detail' (description or details of the action). Returns an empty list if no records are found or if an error occurs during query execution.

Dependencies

• logging
• CDocs.db
• neo4j

Required Imports

import logging
from CDocs import db

Usage Example

# Basic usage - get last 100 activities
activity_logs = get_user_activity()

# Filter by date range
from datetime import datetime, timedelta
date_to = datetime.now().isoformat()
date_from = (datetime.now() - timedelta(days=7)).isoformat()
weekly_logs = get_user_activity(date_from=date_from, date_to=date_to)

# Filter by specific user and action types
user_logs = get_user_activity(
    user_id='user123',
    action_types=['login', 'document_update', 'document_delete'],
    limit=50
)

# Process results
for log in user_logs:
    print(f"{log['timestamp']}: {log['user_name']} performed {log['action_type']}")
    print(f"  Details: {log['action_detail']}")

# Get all activities for a specific date
specific_date = '2024-01-15'
day_logs = get_user_activity(
    date_from=f'{specific_date}T00:00:00',
    date_to=f'{specific_date}T23:59:59',
    limit=1000
)

Best Practices

• Always use ISO format dates (YYYY-MM-DDTHH:MM:SS) for date_from and date_to parameters to ensure proper filtering
• Set appropriate limit values to avoid retrieving excessive records and impacting database performance
• Handle the empty list return value gracefully in calling code, as it can indicate either no results or an error condition
• Check application logs for detailed error messages if the function returns an empty list unexpectedly
• Ensure the database connection is established before calling this function
• Use specific action_types filters when possible to improve query performance
• The function includes debug logging that counts total AuditEvent and Activity nodes - monitor these logs during troubleshooting
• Be aware that the function queries both new (AuditEvent) and legacy (Activity) node structures for backward compatibility
• Consider implementing pagination for large result sets by calling the function multiple times with different date ranges
• The function uses parameterized queries to prevent Cypher injection attacks - do not modify the query construction logic

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function get_audit_trail 79.0% 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_audit_events 70.2% 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 count_audit_events 65.9% 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_user_action 63.5% 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 get_document_audit_trail 58.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