add events table

utils/event_logger.py

@@ -0,0 +1,123 @@
+from flask import request
+from models import Event, EventType, db
+from typing import Optional, Dict, Any, List
+from datetime import datetime
+
+def log_event(
+    event_type: EventType,
+    user_id: int,
+    details: Optional[Dict[str, Any]] = None
+) -> Event:
+    """
+    Log an event in the system.
+    
+    Args:
+        event_type: The type of event from EventType enum
+        user_id: The ID of the user performing the action
+        details: Optional dictionary containing additional event-specific data
+    
+    Returns:
+        The created Event object
+    """
+    event = Event(
+        event_type=event_type.value,
+        user_id=user_id,
+        details=details or {},
+        ip_address=request.remote_addr if request else None,
+        user_agent=request.user_agent.string if request and request.user_agent else None
+    )
+    
+    db.session.add(event)
+    db.session.commit()
+    return event
+
+def get_user_events(
+    user_id: int,
+    event_type: Optional[EventType] = None,
+    start_date: Optional[datetime] = None,
+    end_date: Optional[datetime] = None,
+    limit: int = 100
+) -> List[Event]:
+    """
+    Retrieve events for a specific user with optional filtering.
+    
+    Args:
+        user_id: The ID of the user to get events for
+        event_type: Optional event type to filter by
+        start_date: Optional start date to filter events
+        end_date: Optional end date to filter events
+        limit: Maximum number of events to return
+    
+    Returns:
+        List of Event objects matching the criteria
+    """
+    query = Event.query.filter_by(user_id=user_id)
+    
+    if event_type:
+        query = query.filter_by(event_type=event_type.value)
+    if start_date:
+        query = query.filter(Event.timestamp >= start_date)
+    if end_date:
+        query = query.filter(Event.timestamp <= end_date)
+    
+    return query.order_by(Event.timestamp.desc()).limit(limit).all()
+
+def get_room_events(
+    room_id: int,
+    event_type: Optional[EventType] = None,
+    start_date: Optional[datetime] = None,
+    end_date: Optional[datetime] = None,
+    limit: int = 100
+) -> List[Event]:
+    """
+    Retrieve events related to a specific room with optional filtering.
+    
+    Args:
+        room_id: The ID of the room to get events for
+        event_type: Optional event type to filter by
+        start_date: Optional start date to filter events
+        end_date: Optional end date to filter events
+        limit: Maximum number of events to return
+    
+    Returns:
+        List of Event objects matching the criteria
+    """
+    query = Event.query.filter(Event.details['room_id'].astext.cast(Integer) == room_id)
+    
+    if event_type:
+        query = query.filter_by(event_type=event_type.value)
+    if start_date:
+        query = query.filter(Event.timestamp >= start_date)
+    if end_date:
+        query = query.filter(Event.timestamp <= end_date)
+    
+    return query.order_by(Event.timestamp.desc()).limit(limit).all()
+
+def get_recent_events(
+    event_type: Optional[EventType] = None,
+    start_date: Optional[datetime] = None,
+    end_date: Optional[datetime] = None,
+    limit: int = 100
+) -> List[Event]:
+    """
+    Retrieve recent events across the system with optional filtering.
+    
+    Args:
+        event_type: Optional event type to filter by
+        start_date: Optional start date to filter events
+        end_date: Optional end date to filter events
+        limit: Maximum number of events to return
+    
+    Returns:
+        List of Event objects matching the criteria
+    """
+    query = Event.query
+    
+    if event_type:
+        query = query.filter_by(event_type=event_type.value)
+    if start_date:
+        query = query.filter(Event.timestamp >= start_date)
+    if end_date:
+        query = query.filter(Event.timestamp <= end_date)
+    
+    return query.order_by(Event.timestamp.desc()).limit(limit).all()