123 lines
3.8 KiB
Python
123 lines
3.8 KiB
Python
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() |