Spaces:
Running
Running
File size: 2,645 Bytes
a8a2cf5 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 |
"""
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
|