webapp/app/services/session_service.py

# app/services/session_service.py

from datetime import datetime
from typing import Optional

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, delete

from app.models.session import VasSession
from app.models.user import VasUser


class SessionService:

    @staticmethod
    async def get_user_by_token(
        db: AsyncSession,
        session_id: str
    ) -> Optional[VasUser]:
        result = await db.execute(
            select(VasSession).where(VasSession.id == session_id)
        )
        session_obj = result.scalar_one_or_none()

        if not session_obj:
            return None

        if session_obj.expire_at < datetime.utcnow():
            await SessionService.delete_session(db, session_id)
            return None

        result = await db.execute(
            select(VasUser).where(VasUser.id == session_obj.user_id)
        )
        return result.scalar_one_or_none()

    @staticmethod
    async def delete_session(
        db: AsyncSession,
        session_id: str
    ) -> None:
        await db.execute(
            delete(VasSession).where(VasSession.id == session_id)
        )
        await db.commit()