Sign inorSign up
47 lines
jobs/session_cleaner.py
Filters expired sessions and logs authenticated-user cleanups.
# Session expiry filtering and audit logging.
import logging
from datetime import datetime
from typing import List, Optional
 
logger = logging.getLogger(__name__)
 
# Type alias for a session record dict.
Session = dict  # keys: session_id (str), user_id (Optional[str]), expires_at (datetime)
 
 
def expired_session_ids(sessions: List[Session], cutoff: datetime) -> List[str]:
    """Return IDs of sessions whose expiry falls before cutoff.
 
    Parameters
    ----------
    sessions : list of Session
        All candidate sessions to check.
    cutoff : datetime
        Sessions that expired before this point are included.
 
    Returns
    -------
    list of str
        Session IDs eligible for deletion.
    """
    return [
        s["session_id"]
        for s in sessions
        if s["expires_at"] < cutoff
    ]
 
 
def log_cleanup(sessions: List[Session], cutoff: datetime) -> None:
    """Write an audit entry for each authenticated user whose sessions expired.
 
    Parameters
    ----------
    sessions : list of Session
        All candidate sessions (both expired and live).
    cutoff : datetime
        Sessions that expired before this point are logged.
    """
    for session in sessions:
        if session["expires_at"] < cutoff:
            user_id: Optional[str] = session["user_id"]
            logger.info("session cleaned for user %s", user_id.lower())