🔍 Code Extractor

Retrieves user activity logs from a Neo4j graph database with optional filtering by date range, user ID, action types, and result limit.

Purpose

This function queries a Neo4j database to fetch user activity records, allowing administrators or systems to audit user actions, generate activity reports, and monitor system usage. It supports flexible filtering through multiple optional parameters and returns formatted activity logs with timestamp, username, action type, and action details. The function handles query construction dynamically based on provided filters and includes error handling to return 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.user_id = $user_id")
            params['user_id'] = user_id
            
        if action_types:
            conditions.append("a.action_type 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
        query = f"""
        MATCH (a:Activity)-[:PERFORMED_BY]->(u:User)
        {where_clause}
        RETURN a.timestamp as timestamp, 
               u.username as user_name,
               a.action_type as action_type,
               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)
        
        # 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 included. 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 included. If None, no upper date bound is applied.

user_id: Optional string identifier for a specific user. When provided, only activities performed by this user will be returned. If None, activities from all users are included.

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

action_types: Optional list of strings representing action types to filter by (e.g., ['login', 'document_edit', 'delete']). Only activities matching these types will be returned. If None, all action types are included.

Return Value

Returns a list of dictionaries, where each dictionary represents an activity record with four keys: 'timestamp' (datetime of the activity), 'user_name' (username who performed the action), 'action_type' (type of action performed), and 'action_detail' (additional details about the action). Returns an empty list if an error occurs or no records match the filters.

Dependencies

• logging
• CDocs.db
• CDocs.models.document
• CDocs.models.user_extensions
• CDocs.db.schema_manager
• CDocs.config

Required Imports

import logging
from CDocs import db

Usage Example

# Basic usage - get last 100 activities
from CDocs import db
import logging

logger = logging.getLogger(__name__)

# Get all recent activities
activities = get_user_activity()

# Filter by date range
activities = get_user_activity(
    date_from='2024-01-01T00:00:00',
    date_to='2024-01-31T23:59:59',
    limit=50
)

# Filter by specific user and action types
activities = get_user_activity(
    user_id='user123',
    action_types=['login', 'document_edit', 'document_delete'],
    limit=200
)

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

Best Practices

• Always set an appropriate limit to prevent excessive memory usage when querying large datasets
• Use ISO format dates (YYYY-MM-DDTHH:MM:SS) for date_from and date_to parameters
• Handle the empty list return value which indicates either no results or an error occurred
• Check logs for error messages when empty results are unexpected
• Ensure the database connection is established before calling this function
• Consider pagination for large result sets by adjusting the limit parameter and using date ranges
• Validate action_types list contains valid action type strings that exist in your database
• Be aware that the function uses parameterized queries to prevent injection attacks
• The function returns empty list on exceptions, so implement additional error handling if needed

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function get_user_activity 94.5% 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 68.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 64.0% 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 57.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_users 57.0% similar

  Retrieves user information from a Neo4j graph database for display in an admin panel, with optional filtering by user ID.

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