Spaces:
Running
Running
| """ | |
| Session Tracking Middleware | |
| Tracks user activity and updates session timestamps. | |
| """ | |
| from starlette.middleware.base import BaseHTTPMiddleware | |
| from starlette.requests import Request | |
| from starlette.responses import Response | |
| import logging | |
| logger = logging.getLogger(__name__) | |
| class SessionTrackingMiddleware(BaseHTTPMiddleware): | |
| """ | |
| Middleware to track user session activity. | |
| Updates last_activity timestamp for authenticated requests. | |
| """ | |
| async def dispatch(self, request: Request, call_next): | |
| """ | |
| Process request and track session activity. | |
| Args: | |
| request: Incoming HTTP request | |
| call_next: Next middleware/route handler | |
| Returns: | |
| Response from the route handler | |
| """ | |
| # Process the request | |
| response: Response = await call_next(request) | |
| # Check if user is authenticated (has Authorization header) | |
| auth_header = request.headers.get("Authorization") | |
| if auth_header and auth_header.startswith("Bearer "): | |
| token = auth_header.split(" ")[1] | |
| # Extract user_id from token (if valid) | |
| try: | |
| from ...services.user_service import UserService | |
| from ...db.database import AsyncSessionLocal | |
| # Decode token to get user_id | |
| payload = decode_token(token) | |
| user_id = payload.get("sub") | |
| if user_id: | |
| try: | |
| async with AsyncSessionLocal() as session: | |
| sessions = await UserService.get_active_sessions(int(user_id), session) | |
| if sessions: | |
| latest_session = sessions[0] | |
| await UserService.update_session_activity(latest_session["id"], session) | |
| except Exception as e: | |
| logger.warning(f"Failed to update session activity: {e}") | |
| except Exception as e: | |
| # Token invalid or expired - ignore silently | |
| pass | |
| return response | |
| def decode_token(token: str) -> dict: | |
| """ | |
| Helper function to decode JWT token. | |
| Args: | |
| token: JWT token string | |
| Returns: | |
| Decoded payload dictionary | |
| """ | |
| from jose import jwt | |
| from ...core.config.config import settings | |
| payload = jwt.decode( | |
| token, | |
| settings.SECRET_KEY, | |
| algorithms=["HS256"] | |
| ) | |
| return payload | |