# app/services/session_service.py from datetime import datetime, timedelta import uuid from sqlalchemy.orm import Session as DBSession from sqlalchemy import delete from app.models.session import VasSession from app.models.user import VasUser class SessionService: # ============================ # token → user(鉴权用) # ============================ @staticmethod def get_user_by_token(db: DBSession, session_id: str) -> VasUser: session_obj = db.query(VasSession).filter(VasSession.id == session_id).first() if not session_obj: return None # session 是否过期 if session_obj.expire_at < datetime.utcnow(): # 自动删除过期 session SessionService.delete_session(db, session_id) return None user = db.query(VasUser).filter(VasUser.id == session_obj.user_id).first() return user # ============================ # 删除 session(登出) # ============================ @staticmethod def delete_session(db: DBSession, session_id: str): db.query(VasSession).filter(VasSession.id == session_id).delete() db.commit()