from datetime import datetime, UTC from typing import Optional from sqlalchemy import update from sqlmodel import select from starlette.authentication import AuthenticationBackend, AuthCredentials, BaseUser, UnauthenticatedUser from starlette.requests import HTTPConnection from db.models import Session, User from db.session import get_session_context class AuthBackend(AuthenticationBackend): async def authenticate(self, request: HTTPConnection) -> Optional[tuple[AuthCredentials, BaseUser]]: if "id" not in request.session: return None query = ( select(Session) .join(User) .where( Session.id == request.session['id'] ) ) with get_session_context() as db_session: session: Session = db_session.exec(query).first() if session is not None: db_session.execute( update(Session).where(Session.id == session.id) ) db_session.commit() return ( AuthCredentials( [ "authenticated", *[ connection.type for connection in session.user.app_connections ] ] if session is not None else [] ), session.user if session is not None else UnauthenticatedUser() )