""" Analytics Service - Business Logic for Usage Analytics Following Python Quick Start Guide: - Service layer contains business logic - Orchestrates repository calls - Returns DTOs/dicts for API layer """ from typing import Dict, List, Any from app.repositories.token_repository import TokenRepository class AnalyticsService: """ Analytics service - provides usage analytics and statistics. Responsibilities: - Get active sessions by client - Get usage summary statistics - Transform data for presentation """ def __init__(self, token_repo: TokenRepository = None): """Initialize service with repository.""" self.token_repo = token_repo or TokenRepository() def get_active_sessions(self) -> Dict[str, Any]: """ Get all active sessions with user and client information. Returns: Dict with summary stats and detailed session list """ # Get detailed active tokens active_tokens = self.token_repo.get_active_tokens_by_client() # Get summary by client summary = self.token_repo.get_active_sessions_summary() # Calculate overall stats total_active_users = len(set(token['user_id'] for token in active_tokens)) total_active_tokens = len(active_tokens) total_clients = len(set(token['client_id'] for token in active_tokens if token['client_id'])) return { 'summary': { 'total_active_users': total_active_users, 'total_active_tokens': total_active_tokens, 'total_clients_in_use': total_clients }, 'by_client': summary, 'detailed_sessions': active_tokens } def get_client_usage_stats(self, client_id: str) -> Dict[str, Any]: """ Get usage statistics for a specific client. Args: client_id: The client ID to get stats for Returns: Dict with client usage statistics """ all_sessions = self.get_active_sessions() # Filter for specific client client_sessions = [ session for session in all_sessions['detailed_sessions'] if session['client_id'] == client_id ] unique_users = len(set(session['user_id'] for session in client_sessions)) return { 'client_id': client_id, 'active_users': unique_users, 'active_tokens': len(client_sessions), 'sessions': client_sessions } def get_user_active_clients(self, user_id: int) -> List[Dict[str, Any]]: """ Get all clients that a specific user is currently using. Args: user_id: The user ID to get active clients for Returns: List of client information dicts """ all_sessions = self.get_active_sessions() # Filter for specific user user_sessions = [ session for session in all_sessions['detailed_sessions'] if session['user_id'] == user_id ] # Group by client clients = {} for session in user_sessions: client_id = session['client_id'] if client_id and client_id not in clients: clients[client_id] = { 'client_id': client_id, 'client_name': session['client_name'], 'last_access': session['created_at'], 'expires_at': session['expires_at'] } return list(clients.values()) def get_user_analytics(self, user_id: int) -> Dict[str, Any]: """ Get analytics for a specific user (their own sessions only). Args: user_id: The user ID to get analytics for Returns: Dict with user's session summary and active clients """ # Get all sessions and filter for this user all_sessions = self.get_active_sessions() user_sessions = [ session for session in all_sessions['detailed_sessions'] if session['user_id'] == user_id ] # Count unique clients unique_clients = len(set(s['client_id'] for s in user_sessions if s['client_id'])) return { 'summary': { 'total_active_sessions': len(user_sessions), 'total_clients': unique_clients }, 'active_sessions': user_sessions }