invalidate_all_sessions.py 1.2 KB

12345678910111213141516171819202122232425262728293031323334353637383940
  1. #!/usr/bin/env python3
  2. """
  3. 将会话表 sessions 中所有会话置为失效(用于测试登录等场景)。
  4. 在 backend-client 目录下执行: python scripts/invalidate_all_sessions.py
  5. """
  6. import sys
  7. from pathlib import Path
  8. # 将项目根加入 path
  9. sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
  10. from app.database import SessionLocal
  11. from app.models.session import Session
  12. def main():
  13. from app.utils.history import create_session_history
  14. db = SessionLocal()
  15. try:
  16. # 先写历史表,再改/删:将即将失效的会话写入 sessions_his
  17. sessions_to_invalidate = db.query(Session).filter(Session.is_active == True).all()
  18. for s in sessions_to_invalidate:
  19. create_session_history(db, s, "UPDATE")
  20. updated = db.query(Session).filter(Session.is_active == True).update(
  21. {"is_active": False}, synchronize_session=False
  22. )
  23. db.commit()
  24. print(f"已将会话全部置为失效,影响行数: {updated}")
  25. except Exception as e:
  26. db.rollback()
  27. print(f"执行失败: {e}")
  28. raise
  29. finally:
  30. db.close()
  31. if __name__ == "__main__":
  32. main()