🔍 Code Extractor

Processes a single event record from a database by normalizing timestamps, adding user information, and parsing JSON details fields.

Purpose

This function transforms raw database event records into a standardized format suitable for data analysis and pandas DataFrame operations. It handles timestamp normalization to prevent NaN values, adds username information from the record, and attempts to parse stringified JSON details into dictionary objects. The function is designed to be robust against various timestamp formats and malformed data.

Source Code

def _process_event_record(record: Dict) -> Dict[str, Any]:
    """
    Process a single event record from the database.
    
    Args:
        record: Database record with event data
        
    Returns:
        Processed event dictionary
    """
    event = record['e']
    event['userName'] = record.get('userName')
    
    # Ensure timestamp is properly formatted to prevent NaN in pandas
    if 'timestamp' in event:
        try:
            # Convert to string in ISO format if it's a datetime object
            if isinstance(event['timestamp'], datetime):
                event['timestamp'] = event['timestamp'].isoformat()
            # Ensure it's a string that pandas can parse
            elif event['timestamp'] is None:
                event['timestamp'] = ''
            # Make sure we have a non-empty string
            elif not event['timestamp']:
                event['timestamp'] = ''
            # If it's already a string but not ISO format, try to convert it
            elif isinstance(event['timestamp'], str) and not event['timestamp'].endswith('Z') and 'T' not in event['timestamp']:
                try:
                    dt = datetime.fromisoformat(event['timestamp'])
                    event['timestamp'] = dt.isoformat()
                except:
                    pass  # Keep as is if we can't parse it
        except Exception as e:
            logger.error(f"Error formatting timestamp: {e}")
            # Provide default value to prevent NaN
            event['timestamp'] = ''
    
    # 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
            
    return event

Parameters

Parameter Details

record: A dictionary containing database event data. Expected to have an 'e' key with the event object, and optionally a 'userName' key. The event object may contain 'timestamp' (datetime, string, or None) and 'details' (string or dict) fields.

Return Value

Type: Dict[str, Any]

Returns a dictionary representing the processed event with normalized fields. The dictionary includes the original event data with modifications: 'userName' added from the record, 'timestamp' converted to ISO format string (or empty string if invalid/missing), and 'details' parsed from JSON string to dictionary if applicable. Type: Dict[str, Any]

Dependencies

• logging
• datetime
• json

Required Imports

from typing import Dict, Any
from datetime import datetime
import logging

Conditional/Optional Imports

These imports are only needed under specific conditions:

import json

Usage Example

from datetime import datetime
from typing import Dict, Any
import logging

logger = logging.getLogger(__name__)

def _process_event_record(record: Dict) -> Dict[str, Any]:
    # ... function code ...
    pass

# Example usage
db_record = {
    'e': {
        'eventId': '123',
        'eventType': 'login',
        'timestamp': datetime(2024, 1, 15, 10, 30, 0),
        'details': '{"ip": "192.168.1.1", "browser": "Chrome"}'
    },
    'userName': 'john.doe'
}

processed_event = _process_event_record(db_record)
print(processed_event)
# Output: {'eventId': '123', 'eventType': 'login', 'timestamp': '2024-01-15T10:30:00', 'details': {'ip': '192.168.1.1', 'browser': 'Chrome'}, 'userName': 'john.doe'}

Best Practices

• Ensure a logger instance is configured before calling this function to capture timestamp formatting errors
• The function is defensive and handles various timestamp formats (datetime objects, ISO strings, non-ISO strings, None, empty strings)
• Returns empty string for invalid timestamps to prevent NaN values in pandas DataFrames
• JSON parsing of 'details' field fails silently, keeping the original string value if parsing fails
• The function modifies the event dictionary in-place before returning it
• Suitable for use in batch processing pipelines with map/apply operations
• Consider validating the structure of the input record dictionary before calling this function in production code

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function log_event 47.2% 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 log_review_event 44.6% 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 prepare_audit_data_for_document_processor 42.9% similar

  Prepares comprehensive audit data for a controlled document version, aggregating information from document history, reviews, approvals, and audit events into a structured format for DocumentProcessor.

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

• function migrate_audit_trail 42.4% similar

  Migrates existing audit events from a legacy format to a new AuditTrail structure in a Neo4j database, returning migration statistics.

  From: /tf/active/vicechatdev/CDocs/db/db_operations.py

• function migrate_audit_events 42.1% similar

  Migrates orphaned AuditEvent nodes from CDocs nodes to an AuditTrail node in a Neo4j graph database by deleting old relationships and creating new ones.

  From: /tf/active/vicechatdev/CDocs/db/schema_manager.py