72 lines
2.7 KiB
Python
72 lines
2.7 KiB
Python
from flask import request
|
|
from models import Event, EventType, db
|
|
from typing import Optional, Dict, Any, List
|
|
from datetime import datetime
|
|
from flask_login import current_user
|
|
from sqlalchemy import desc
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def log_event(event_type: str, details: Optional[Dict[str, Any]] = None, user_id: Optional[int] = None) -> Event:
|
|
"""
|
|
Log an event to the database.
|
|
|
|
Args:
|
|
event_type: The type of event (must match EventType enum)
|
|
details: Optional dictionary containing event details
|
|
user_id: Optional user ID (defaults to current user)
|
|
|
|
Returns:
|
|
The created Event object
|
|
"""
|
|
logger.debug(f"Creating event of type: {event_type}")
|
|
logger.debug(f"Event details: {details}")
|
|
|
|
if user_id is None and current_user.is_authenticated:
|
|
user_id = current_user.id
|
|
logger.debug(f"Using current user ID: {user_id}")
|
|
|
|
try:
|
|
event = Event(
|
|
event_type=event_type,
|
|
user_id=user_id,
|
|
timestamp=datetime.utcnow(),
|
|
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
|
|
)
|
|
|
|
logger.debug(f"Created event object: {event}")
|
|
db.session.add(event)
|
|
# Don't commit here - let the caller handle the transaction
|
|
logger.debug("Event object added to session")
|
|
return event
|
|
except Exception as e:
|
|
logger.error(f"Error creating event: {str(e)}")
|
|
raise
|
|
|
|
def get_user_events(user_id: int, limit: int = 50) -> List[Event]:
|
|
"""Get recent events for a specific user"""
|
|
return Event.query.filter_by(user_id=user_id).order_by(desc(Event.timestamp)).limit(limit).all()
|
|
|
|
def get_room_events(room_id: int, limit: int = 50) -> List[Event]:
|
|
"""Get recent events for a specific room"""
|
|
return Event.query.filter(
|
|
Event.details['room_id'].astext == str(room_id)
|
|
).order_by(desc(Event.timestamp)).limit(limit).all()
|
|
|
|
def get_recent_events(limit: int = 50) -> List[Event]:
|
|
"""Get most recent events across all types"""
|
|
return Event.query.order_by(desc(Event.timestamp)).limit(limit).all()
|
|
|
|
def get_events_by_type(event_type: str, limit: int = 50) -> List[Event]:
|
|
"""Get recent events of a specific type"""
|
|
return Event.query.filter_by(event_type=event_type).order_by(desc(Event.timestamp)).limit(limit).all()
|
|
|
|
def get_events_by_date_range(start_date: datetime, end_date: datetime, limit: int = 50) -> List[Event]:
|
|
"""Get events within a date range"""
|
|
return Event.query.filter(
|
|
Event.timestamp >= start_date,
|
|
Event.timestamp <= end_date
|
|
).order_by(desc(Event.timestamp)).limit(limit).all() |